mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-12 14:38:49 -10:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c425192836 | ||
|
|
63e09ba794 | ||
|
|
451a02e133 | ||
|
|
84726a2c6d |
No files matched your search
@@ -235,6 +235,8 @@ one is dropped without a message.
|
||||
|
||||
### Readiness, startup proof and failure
|
||||
|
||||
Managed starts share [memory admission and pacing](host-resources.md#pace-managed-startup-memory) within the owning daemon. A queued request has status/cancel/continue controls; a timed-out client does not authorize another native launch.
|
||||
|
||||
**Readiness.** After launch OpenRig polls the pane until the harness is interactive (backing off 1, 2, 4, 8 and 16
|
||||
seconds), for up to 30 seconds by default. Change the window with
|
||||
`rig config set runtime.readiness_timeout_seconds <1-600>` (or `OPENRIG_RUNTIME_READINESS_TIMEOUT_SECONDS`); it
|
||||
|
||||
@@ -59,3 +59,65 @@ and is retried; normal config
|
||||
writes reject invalid values. Changing `transcripts.enabled` or
|
||||
`transcripts.path` still requires a daemon restart. This feature does not impose
|
||||
CPU quotas or allocate build slots.
|
||||
|
||||
|
||||
## Pace managed startup memory
|
||||
|
||||
One shared admission gate coordinates this daemon's managed starts and resumes,
|
||||
including overlapping rig requests, staged successors and in-place restarts.
|
||||
It admits one at a time by default. It preserves the selected conversation,
|
||||
model, effort and dependency order; it does not evict healthy seats or choose a
|
||||
fresh conversation to reduce memory. It does not coordinate other daemons,
|
||||
users, manual native launches or memory on remote execution hosts. A local
|
||||
proxy's admission is not protection for its remote guest's memory.
|
||||
|
||||
Readiness proves the runtime became interactive; it does not prove its memory
|
||||
peak ended. Admission retains the slot for bounded pacing after readiness.
|
||||
Linux observations use MemAvailable, readable cgroup v2 headroom and memory PSI
|
||||
(pressure is high at some avg10 >= 10% or full avg10 >= 1%). macOS uses the
|
||||
read-only memory_pressure -Q percentage and the kernel's memory-pressure level.
|
||||
These are OS-specific observations, not equivalent guarantees of free capacity.
|
||||
Samples older than five seconds are unavailable.
|
||||
|
||||
Set these environment variables on the daemon's next ordinary start; they do
|
||||
not change a running daemon's policy:
|
||||
|
||||
| Variable | Default | Meaning |
|
||||
| --- | --- | --- |
|
||||
| OPENRIG_STARTUP_CONCURRENCY | 1 | Concurrent managed startup operations, integer 1-16. |
|
||||
| OPENRIG_STARTUP_HEADROOM_PERCENT | 25 | Defer below this observed headroom percentage, integer 0-90. |
|
||||
| OPENRIG_STARTUP_SETTLE_MS | 3000 | Pacing after the native operation returns, 0-60000 ms, capped by SETTLE_LIMIT_MS. |
|
||||
| OPENRIG_STARTUP_SETTLE_LIMIT_MS | 15000 | Maximum settling time after native return, 0-120000 ms. |
|
||||
| OPENRIG_STARTUP_STRICT_TELEMETRY | unset | Set to 1 to wait when memory observations are unavailable. |
|
||||
|
||||
The defaults are conservative pacing settings, not a promise that a particular
|
||||
conversation fits a laptop. Missing, stale or unreadable observations normally
|
||||
serialize starts, retain readiness and bounded pacing, and report that memory
|
||||
protection is unavailable. Strict telemetry waiting is opt-in. Known pressure
|
||||
or a real headroom deficit can defer dispatch even with other telemetry missing.
|
||||
A one-seat deficit means no startup operation holds another slot: waiting for
|
||||
another start to finish cannot by itself make that seat fit.
|
||||
|
||||
Inspect and control the original request without submitting another launch:
|
||||
|
||||
```bash
|
||||
rig launch my-rig my-seat --startup-admission status --json
|
||||
rig launch my-rig my-seat --startup-admission cancel --json
|
||||
rig launch my-rig my-seat --startup-admission continue --json
|
||||
```
|
||||
|
||||
The rig may be an ID or an unambiguous name; --host selects the destination
|
||||
daemon. Cancel applies only before native dispatch or an incumbent stop. It
|
||||
never kills a native child. Continue deliberately admits one queued request
|
||||
despite its observed memory risk, once its concurrency slot is free; it cannot
|
||||
repeat an unresolved launch. Admission facts also appear on node inventory,
|
||||
node detail and node.startup_admission events. requestActive distinguishes a
|
||||
live waiting request from retained historical status after a daemon restart.
|
||||
|
||||
A client timeout does not cancel server-side startup. Repeated matching requests
|
||||
join its existing attempt while it remains owned. After a daemon restart,
|
||||
reconcile an interrupted native attempt before retrying; a live or unknown
|
||||
process is not permission to send another resume. A queued request that never
|
||||
dispatched can be requested again through the original start command. Retained
|
||||
ready evidence or proven native absence resolves interrupted custody. Failed
|
||||
or unconfirmed starts remain visible rather than being retried automatically.
|
||||
@@ -82,18 +82,43 @@ export function launchCommand(depsOverride?: StatusDeps): Command {
|
||||
.argument("[nodeRef]", "Node logical ID or node ID (single target)")
|
||||
.option("--seats <ids>", "Comma-separated logical IDs for subset launch")
|
||||
.option("--hold-reason <reason>", "Reason for holding non-target seats")
|
||||
.option("--startup-admission <action>", "Inspect a pending start, cancel before dispatch, or continue one despite memory risk: status|cancel|continue")
|
||||
.option("--snapshot-id <id>", "Use this exact restore-usable snapshot")
|
||||
.option("--retry-startup-from <member-file>", "Retry a stopped first start that failed during projection, using its original member fragment")
|
||||
.option("--rig-root <path>", "Original absolute source root for --retry-startup-from")
|
||||
.option("--plan", "Show restore selection and non-target effects without mutation; does not preview first start of a never-started materialized seat")
|
||||
.option("--json", "JSON output")
|
||||
.option("--host <id>", "Run on a remote host declared in ~/.openrig/hosts.yaml")
|
||||
.action(async (rigHandle: string, nodeRef: string | undefined, opts: { json?: boolean; holdReason?: string; seats?: string; host?: string; snapshotId?: string; plan?: boolean; retryStartupFrom?: string; rigRoot?: string }) => {
|
||||
.action(async (rigHandle: string, nodeRef: string | undefined, opts: { startupAdmission?: string; json?: boolean; holdReason?: string; seats?: string; host?: string; snapshotId?: string; plan?: boolean; retryStartupFrom?: string; rigRoot?: string }) => {
|
||||
// OPR.0.4.6.MH1 FR-2: selected-host routing — explicit --host wins;
|
||||
// else the persisted selection feeds the SHIPPED --host path; no
|
||||
// selection = today exactly.
|
||||
opts.host = resolveEffectiveHost(opts.host);
|
||||
const deps = getDeps();
|
||||
if (opts.startupAdmission) {
|
||||
if (!nodeRef || !["status", "cancel", "continue"].includes(opts.startupAdmission) || opts.seats || opts.plan || opts.snapshotId || opts.retryStartupFrom || opts.rigRoot || opts.holdReason) {
|
||||
console.error("Startup admission requires one seat and status, cancel or continue; do not combine it with a launch selection.");
|
||||
process.exitCode = 1; return;
|
||||
}
|
||||
const body = { action: opts.startupAdmission };
|
||||
if (opts.host) {
|
||||
const { runRemoteHttpOp, lookupRemoteRigId } = await import("../remote-host-ops.js");
|
||||
const lookup = await lookupRemoteRigId(opts.host, rigHandle, deps);
|
||||
if (lookup.kind === "ambiguous") { refuseAmbiguousRig(rigHandle, lookup.ids, opts.json, opts.host, lookup.count); return; }
|
||||
const id = lookup.kind === "resolved" ? lookup.rigId : rigHandle;
|
||||
const answer = await runRemoteHttpOp(opts.host, "POST", `/api/rigs/${encodeURIComponent(id)}/nodes/${encodeURIComponent(nodeRef)}/startup-admission`, body, deps, opts);
|
||||
console.log(JSON.stringify(answer.ok ? answer.data : answer, null, 2));
|
||||
if (!answer.ok) process.exitCode = 1;
|
||||
} else {
|
||||
const client = await getClient(deps); if (!client) { process.exitCode = 1; return; }
|
||||
const lookup = await resolveRigHandle(client, rigHandle);
|
||||
if (lookup.kind === "ambiguous") { refuseAmbiguousRig(rigHandle, lookup.ids, opts.json); return; }
|
||||
const id = lookup.kind === "resolved" ? lookup.id : rigHandle;
|
||||
const answer = await client.post(`/api/rigs/${encodeURIComponent(id)}/nodes/${encodeURIComponent(nodeRef)}/startup-admission`, body);
|
||||
console.log(JSON.stringify(answer.data, null, 2)); if (answer.status >= 400) process.exitCode = 1;
|
||||
}
|
||||
return;
|
||||
}
|
||||
let retryBody: { retryStartupFrom: { member: Record<string, unknown>; rigRoot: string } } | undefined;
|
||||
if (opts.retryStartupFrom || opts.rigRoot) {
|
||||
if (!nodeRef || !opts.retryStartupFrom || !opts.rigRoot || !isAbsolute(opts.rigRoot) || opts.seats || opts.snapshotId || opts.plan || opts.holdReason) {
|
||||
|
||||
@@ -54,6 +54,13 @@ describe("rig launch --seats", () => {
|
||||
process.exitCode = undefined;
|
||||
});
|
||||
|
||||
it.each(["status", "cancel", "continue"])("routes startup admission %s without posting a new launch", async action => {
|
||||
const deps = makeDeps({ "startup-admission": { status: 200, data: { ok: true, admission: { phase: "held" } } } });
|
||||
await launchCommand(deps).parseAsync(["node", "rig", "rig-1", "dev.driver", "--startup-admission", action]);
|
||||
expect(deps._client.post).toHaveBeenCalledTimes(1);
|
||||
expect(deps._client.post).toHaveBeenCalledWith("/api/rigs/rig-1/nodes/dev.driver/startup-admission", { action });
|
||||
});
|
||||
|
||||
it("posts to launch-subset with seats array and holdReason", async () => {
|
||||
const deps = makeDeps({
|
||||
"launch-subset": {
|
||||
|
||||
@@ -379,6 +379,8 @@ function usableEnvLine(lines: readonly string[], name: string): boolean {
|
||||
}
|
||||
|
||||
export class TmuxAdapter {
|
||||
/** Shared owning-daemon admission, absent on isolated adapters and test fixtures. */
|
||||
startupAdmission?: import("../domain/startup-admission.js").StartupAdmission;
|
||||
/** Read for each new agent/successor. Unconfigured adapters retain legacy behavior. */
|
||||
launchEnvironment?: () => { inheritTmuxEnvironment: boolean; env: Readonly<Record<string, string | undefined>>;
|
||||
/** The runtime directory a seat gets when it has none of its own (seat-runtime-dir.ts); none on macOS. */
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
import type Database from "better-sqlite3";
|
||||
import type { EventBus } from "./event-bus.js";
|
||||
import type { SessionRegistry } from "./session-registry.js";
|
||||
import type { TmuxAdapter } from "../adapters/tmux.js";
|
||||
import { StartupAdmission, type StartupAdmissionSnapshot } from "./startup-admission.js";
|
||||
import { observeStartupMemory } from "./startup-memory.js";
|
||||
import { verifyPaneIdleShell } from "./native-process-lineage.js";
|
||||
|
||||
/** Install once on the shared adapter; nested launch/restore services must reuse this instance. */
|
||||
export function daemonStartupAdmission(db: Database.Database, events: EventBus, registry: SessionRegistry, tmux: TmuxAdapter): StartupAdmission {
|
||||
return new StartupAdmission({
|
||||
observe: observeStartupMemory,
|
||||
ownsLifecycle: nodeId => tmux.deliveryGuard?.ownsLifecycle(nodeId) ?? false,
|
||||
lifecycle: (nodeId, work) => tmux.deliveryGuard ? tmux.deliveryGuard.lifecycle([nodeId], work) : work(),
|
||||
selection: nodeId => JSON.stringify({ generation: registry.currentOccupantTenure(nodeId)?.generationUuid ?? null,
|
||||
node: db.prepare("SELECT runtime,model,effort,effort_choice,cwd,codex_config_profile FROM nodes WHERE id=?").get(nodeId),
|
||||
binding: registry.getBindingForNode(nodeId) }),
|
||||
history: () => {
|
||||
const rows = db.prepare(`SELECT e.seq,e.payload FROM events e JOIN
|
||||
(SELECT node_id,MAX(seq) seq FROM events WHERE type='node.startup_admission' GROUP BY node_id) l ON e.seq=l.seq`).all() as Array<{seq: number; payload: string}>;
|
||||
return rows.map(row => {
|
||||
const snapshot = (JSON.parse(row.payload) as { admission: StartupAdmissionSnapshot }).admission;
|
||||
return { ...snapshot, ...(snapshot.phase === "starting" ? { dispatchSeq: row.seq } : {}) };
|
||||
});
|
||||
},
|
||||
persist: admission => {
|
||||
const node = db.prepare("SELECT rig_id FROM nodes WHERE id=?").get(admission.nodeId) as { rig_id: string } | undefined;
|
||||
if (!node) throw new Error("The startup seat no longer exists");
|
||||
return events.emit({ type: "node.startup_admission", rigId: node.rig_id, nodeId: admission.nodeId, admission }).seq;
|
||||
},
|
||||
reconcile: async snapshot => {
|
||||
const binding = registry.getBindingForNode(snapshot.nodeId);
|
||||
const targets = snapshot.nativeTargets?.length ? snapshot.nativeTargets : [snapshot.nativeTarget ?? binding?.tmuxSession].filter((value): value is string => Boolean(value));
|
||||
if (!targets.length) return "unknown";
|
||||
// Ready evidence must be newer than this actual dispatch, never the previous occupant's ready bit.
|
||||
if (snapshot.dispatchSeq !== undefined) {
|
||||
const ready = db.prepare("SELECT 1 FROM events WHERE node_id=? AND type='node.startup_ready' AND seq>? LIMIT 1")
|
||||
.get(snapshot.nodeId, snapshot.dispatchSeq);
|
||||
if (ready && !snapshot.kind.includes("successor") && snapshot.kind !== "in-place restart") return "ready";
|
||||
}
|
||||
if (tmux.probeSession) {
|
||||
for (const target of targets) {
|
||||
const probe = await tmux.probeSession(target);
|
||||
if (probe.state === "absent") continue;
|
||||
if (probe.state !== "present") return "unknown";
|
||||
if (!await verifyPaneIdleShell({ target, tmux })) return "live";
|
||||
}
|
||||
return "absent";
|
||||
}
|
||||
return "unknown";
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -380,6 +380,7 @@ export class NodeLauncher {
|
||||
}
|
||||
// The guard resolves the selected seat before writing to its immutable pane id.
|
||||
const inputTarget = this.tmuxAdapter.deliveryGuard ? nodeId : pane;
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(nodeId, sessionName);
|
||||
const ended = await terminatePaneInPlace(this.tmuxAdapter, pane, this.paneExit, "The seat's process", inputTarget);
|
||||
if (!ended.ok) return { ok: false, code: "pane_process_not_ended", message: ended.message };
|
||||
const shell = (await this.tmuxAdapter.getDefaultShell()) ?? "/bin/sh";
|
||||
|
||||
@@ -938,6 +938,7 @@ export class RestoreOrchestrator {
|
||||
const afterProbe = current();
|
||||
if (!afterProbe || afterProbe.name !== row.name || afterProbe.pane !== row.pane) return "Startup occupant or correction changed during the terminal probe; no replacement attempted.";
|
||||
this.tmuxAdapter.deliveryGuard?.checkInput(row.name);
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(target.nodeId, row.name);
|
||||
const killed = await this.tmuxAdapter.killSession(row.name);
|
||||
if (!killed.ok) return "Could not stop the startup occupant; no replacement attempted.";
|
||||
if (!current()) return "Startup occupant changed during stop; no replacement launched.";
|
||||
@@ -1083,6 +1084,15 @@ export class RestoreOrchestrator {
|
||||
warnings?: string[],
|
||||
preserveStoppedCurrent = false,
|
||||
): Promise<RestoreNodeResult> {
|
||||
const admission = this.tmuxAdapter.startupAdmission;
|
||||
if (admission && !admission.owns(entry.node.id)) {
|
||||
try { return await admission.run(entry.node.id, "restore", () =>
|
||||
this.restoreNodeWithCompensation(entry, rigId, snapshotId, data, opts, warnings, preserveStoppedCurrent), JSON.stringify([snapshotId, opts?.freshLogicalIds, opts?.keepPane, opts?.keepEffort, opts?.selectedHistory?.historySessionId, preserveStoppedCurrent])); }
|
||||
catch (error) { return { nodeId: entry.node.id, logicalId: entry.node.logicalId, status: "awaiting-decision", error: (error as Error).message }; }
|
||||
}
|
||||
const guard = this.tmuxAdapter.deliveryGuard;
|
||||
if (guard && !guard.ownsLifecycle(entry.node.id)) return guard.lifecycle([entry.node.id], () =>
|
||||
this.restoreNodeWithCompensation(entry, rigId, snapshotId, data, opts, warnings, preserveStoppedCurrent));
|
||||
// Compare against the live logical target without changing saved post-launch inputs.
|
||||
const node = entry.node;
|
||||
// Activity-hook restore gaps (#1530) are collected out-of-band: postLaunchRestore
|
||||
@@ -2411,6 +2421,7 @@ export class RestoreOrchestrator {
|
||||
...(claudeExecutable ? [claudeExecutable] : [])] : tail;
|
||||
const mcpTail = claudeMcp === undefined ? guardedTail : [...guardedTail, ...Array(Math.max(0, 9 - guardedTail.length)).fill(undefined), claudeMcp];
|
||||
if (beforeNative && !beforeNative()) return { kind: "attention_required", message: "The selected seat, settings or history changed during native preparation; resume was not dispatched." };
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(nodeId, sessionName);
|
||||
const result = await this.claudeResume.resume(sessionName, resumeType, resumeToken, cwd, resolvedPosture, model, permissionMode, nodeId,
|
||||
...(mcpTail as [...typeof launchTail, advisor?: string, launchAdditions?: SeatLaunchAdditions, claudePermissions?: import("./claude-permission-rules.js").ClaudePermissionRules, beforeNativeDispatch?: () => boolean, executable?: string, claudeMcp?: ClaudeMcp]));
|
||||
if (result.ok) {
|
||||
@@ -2446,6 +2457,7 @@ export class RestoreOrchestrator {
|
||||
: launchTail;
|
||||
const guardedCodexTail = beforeNative ? [...codexTail, ...Array(Math.max(0, 6 - codexTail.length)).fill(undefined), beforeNative] : codexTail;
|
||||
if (beforeNative && !beforeNative()) return { kind: "attention_required", message: "The selected seat, settings or history changed during native preparation; resume was not dispatched." };
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(nodeId, sessionName);
|
||||
const result = await this.codexResume.resume(sessionName, resumeType, resumeToken, cwd, codexConfigProfile, resolvedPosture, model,
|
||||
...(guardedCodexTail as [...typeof launchTail, seatCodexHome?: string, codexPermissions?: import("./codex-member-permissions.js").CodexMemberPermissions, beforeNativeDispatch?: () => boolean]));
|
||||
if (result.ok) {
|
||||
@@ -2473,6 +2485,7 @@ export class RestoreOrchestrator {
|
||||
// session file returns retry_fresh, which the caller maps to the
|
||||
// awaiting-decision stop-and-ask — never a silent fresh start (BR-6).
|
||||
if (this.piResume?.canResume(resumeType, resumeToken)) {
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(nodeId, sessionName);
|
||||
const result = await this.piResume.resume(sessionName, resumeType, resumeToken, cwd, model, resolvedPosture, effort);
|
||||
if (result.ok) {
|
||||
if (result.appliedLaunch && launchGeneration) this.appliedLaunchStore.recordGeneration(launchGeneration, result.appliedLaunch);
|
||||
@@ -2489,6 +2502,7 @@ export class RestoreOrchestrator {
|
||||
return { kind: "failed", message: result.message };
|
||||
}
|
||||
if (this.ompResume?.canResume(resumeType, resumeToken)) {
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(nodeId, sessionName);
|
||||
const result = await this.ompResume.resume(sessionName, resumeType, resumeToken, cwd, model, resolvedPosture);
|
||||
if (result.ok) {
|
||||
if (result.appliedLaunch && launchGeneration) this.appliedLaunchStore.recordGeneration(launchGeneration, result.appliedLaunch);
|
||||
|
||||
@@ -2359,6 +2359,11 @@ export class PodRigInstantiator {
|
||||
* start recorded rather than read an empty request as "none". */
|
||||
keepRecordedAdditions?: boolean;
|
||||
}): Promise<{ status: "launched" | "failed" | "attention_required"; code?: string; error?: string; evidence?: string; sessionName?: string; warnings?: string[] }> {
|
||||
const admission = this.deps.tmuxAdapter?.startupAdmission;
|
||||
if (admission && !admission.owns(input.nodeId)) {
|
||||
try { return await admission.run(input.nodeId, "first start", () => this.launchExistingAgentMember(input)); }
|
||||
catch (error) { return { status: "attention_required", error: (error as Error).message }; }
|
||||
}
|
||||
const guard = this.deps.tmuxAdapter?.deliveryGuard;
|
||||
if (guard && !guard.ownsLifecycle(input.nodeId)) {
|
||||
return guard.lifecycle([input.nodeId], () => this.launchExistingAgentMember(input));
|
||||
|
||||
@@ -1510,6 +1510,11 @@ export class SeatLifecycleService {
|
||||
}
|
||||
const resolved = this.resolveSeat(input.seatRef);
|
||||
if ("code" in resolved) return resolved;
|
||||
const admission = this.tmuxAdapter.startupAdmission;
|
||||
if (admission && !admission.owns(resolved.nodeId)) {
|
||||
try { return await admission.run(resolved.nodeId, "fresh launch", () => this.launchFresh(input), JSON.stringify([input.fresh, input.stop, input.reason, input.operator])); }
|
||||
catch (error) { return { ok: false, code: "attention_required", status: "attention_required", message: (error as Error).message }; }
|
||||
}
|
||||
const guard = this.tmuxAdapter.deliveryGuard;
|
||||
if (guard && !guard.ownsLifecycle(resolved.nodeId)) {
|
||||
return guard.lifecycle([resolved.nodeId], () => this.launchFresh(input));
|
||||
@@ -1567,6 +1572,7 @@ export class SeatLifecycleService {
|
||||
if (canonicalProbe.state === "present") {
|
||||
const currentSession = await this.validateFreshCollision(resolved, canonicalSessionName, input.stop === true);
|
||||
if ("code" in currentSession) return currentSession;
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(resolved.nodeId, canonicalSessionName);
|
||||
const stopped = await this.stopManagedTmuxSeat(
|
||||
resolved,
|
||||
currentSession,
|
||||
|
||||
@@ -37,7 +37,7 @@ import { CODEX_NETWORK_DEFAULT_OVERRIDE } from "./codex-network-default.js";
|
||||
export type InPlaceRestartTmux = Pick<TmuxAdapter,
|
||||
| "getPanePid" | "getPaneCommand" | "listPanes" | "getSessionEnv" | "signalPid" | "setRemainOnExit"
|
||||
| "signalPaneProcess" | "isPaneDead" | "respawnPane" | "getPaneLocalOption" | "restorePaneOption"
|
||||
| "capturePaneHistory" | "capturePaneHistoryParts">;
|
||||
| "capturePaneHistory" | "capturePaneHistoryParts" | "startupAdmission">;
|
||||
|
||||
/** The Codex runtime adapter's resume surface this operation reuses. */
|
||||
export interface InPlaceCodexLauncher {
|
||||
@@ -104,7 +104,7 @@ export interface InPlaceRestartTarget {
|
||||
|
||||
export type InPlaceRestartRefusalCode =
|
||||
| "pane_unverified" | "pane_shape_unknown" | "resume_binding_unverified" | "settings_unverified"
|
||||
| "settings_drift" | "restart_in_progress" | "relaunch_unavailable" | "tmux_probe_failed";
|
||||
| "startup_admission" | "settings_drift" | "restart_in_progress" | "relaunch_unavailable" | "tmux_probe_failed";
|
||||
|
||||
export interface InPlaceRestartRefusal { ok: false; code: InPlaceRestartRefusalCode; message: string; guidance?: string; detail?: unknown }
|
||||
|
||||
@@ -239,10 +239,16 @@ export class CodexInPlaceRestart {
|
||||
}
|
||||
|
||||
async execute(target: InPlaceRestartTarget, input: { reason: string; operator: string | null; seenStartedSeq: number }): Promise<InPlaceRestartReport | InPlaceRestartRefusal> {
|
||||
const admission = this.deps.tmux.startupAdmission;
|
||||
if (admission && !admission.owns(target.seat.nodeId)) {
|
||||
try { return await admission.run(target.seat.nodeId, "in-place restart", () => this.execute(target, input), JSON.stringify([input.reason, input.operator, input.seenStartedSeq])); }
|
||||
catch (error) { return { ok: false, code: "startup_admission", message: (error as Error).message }; }
|
||||
}
|
||||
const observed = await this.observe(target, input.seenStartedSeq);
|
||||
if (!observed.ok) return observed;
|
||||
const { report } = observed;
|
||||
const { sessionName, seat } = target;
|
||||
this.deps.tmux.startupAdmission?.beforeDispatch(target.seat.nodeId, target.sessionName);
|
||||
this.persist({
|
||||
type: RESTART_STARTED, rigId: seat.rigId, nodeId: seat.nodeId, logicalId: seat.logicalId, sessionName,
|
||||
pane: report.pane, paneShape: report.paneShape, oldPid: observed.codexPid, oldStartedAt: observed.codexStartedAt,
|
||||
|
||||
@@ -0,0 +1,321 @@
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import type { StartupMemory } from "./startup-memory.js";
|
||||
|
||||
export type AdmissionPhase = "queued" | "held" | "preparing" | "starting" | "settling" | "complete" | "cancelled" | "uncertain";
|
||||
export interface StartupAdmissionSnapshot {
|
||||
attemptId: string;
|
||||
nodeId: string;
|
||||
kind: string;
|
||||
selection: string;
|
||||
requestKey?: string;
|
||||
phase: AdmissionPhase;
|
||||
reason: string;
|
||||
observedAt: number;
|
||||
dispatchedAt?: number;
|
||||
dispatchSeq?: number;
|
||||
nativeTarget?: string;
|
||||
nativeTargets?: string[];
|
||||
memory?: StartupMemory;
|
||||
protectionUnavailable?: boolean;
|
||||
}
|
||||
|
||||
export interface StartupAdmissionOptions {
|
||||
concurrency: number;
|
||||
minimumAvailablePercent: number;
|
||||
settleMs: number;
|
||||
settleLimitMs: number;
|
||||
staleMs: number;
|
||||
strictTelemetry: boolean;
|
||||
}
|
||||
|
||||
export function startupAdmissionOptions(env: NodeJS.ProcessEnv = process.env): StartupAdmissionOptions {
|
||||
const number = (key: string, fallback: number, min: number, max: number) => {
|
||||
const raw = env[key];
|
||||
if (raw === undefined) return fallback;
|
||||
const n = Number(raw);
|
||||
if (!Number.isFinite(n) || !Number.isInteger(n) || n < min || n > max) throw new Error(`${key} must be an integer in [${min}, ${max}]`);
|
||||
return n;
|
||||
};
|
||||
return {
|
||||
concurrency: number("OPENRIG_STARTUP_CONCURRENCY", 1, 1, 16),
|
||||
minimumAvailablePercent: number("OPENRIG_STARTUP_HEADROOM_PERCENT", 25, 0, 90),
|
||||
settleMs: number("OPENRIG_STARTUP_SETTLE_MS", 3000, 0, 60000),
|
||||
settleLimitMs: number("OPENRIG_STARTUP_SETTLE_LIMIT_MS", 15000, 0, 120000),
|
||||
staleMs: 5000,
|
||||
strictTelemetry: env.OPENRIG_STARTUP_STRICT_TELEMETRY === "1",
|
||||
};
|
||||
}
|
||||
|
||||
export class StartupAdmissionError extends Error {
|
||||
constructor(readonly code: string, message: string) { super(message); }
|
||||
}
|
||||
|
||||
interface Entry {
|
||||
snapshot: StartupAdmissionSnapshot;
|
||||
work: () => Promise<unknown>;
|
||||
resolve: (result: unknown) => void;
|
||||
reject: (error: unknown) => void;
|
||||
result: Promise<unknown>;
|
||||
continueOne: boolean;
|
||||
cancelled?: boolean;
|
||||
waitingForLifecycle?: boolean;
|
||||
}
|
||||
|
||||
export interface StartupAdmissionDeps {
|
||||
observe: () => Promise<StartupMemory>;
|
||||
selection: (nodeId: string) => string;
|
||||
persist: (snapshot: StartupAdmissionSnapshot) => number | void;
|
||||
history: () => StartupAdmissionSnapshot[];
|
||||
/** Read only: a lost reply never authorizes another native launch. */
|
||||
reconcile: (snapshot: StartupAdmissionSnapshot) => Promise<"absent" | "ready" | "live" | "unknown">;
|
||||
/** Register/deduplicate first, then take the existing seat lease before consuming a global slot. */
|
||||
lifecycle?: <T>(nodeId: string, work: () => Promise<T>) => Promise<T>;
|
||||
ownsLifecycle?: (nodeId: string) => boolean;
|
||||
now?: () => number;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
}
|
||||
|
||||
/** One owning daemon's managed native starts, not a host-wide scheduler or a resident-seat quota. */
|
||||
export class StartupAdmission {
|
||||
private readonly scope = new AsyncLocalStorage<ReadonlySet<string>>();
|
||||
private readonly entries = new Map<string, Entry>();
|
||||
private readonly queue: Entry[] = [];
|
||||
private readonly unresolved = new Map<string, StartupAdmissionSnapshot>();
|
||||
private readonly latest = new Map<string, StartupAdmissionSnapshot>();
|
||||
private readonly lastPersistedAt = new Map<string, number>();
|
||||
private active = 0;
|
||||
private pumping = false;
|
||||
private timer?: ReturnType<typeof setTimeout>;
|
||||
private readonly now: () => number;
|
||||
private readonly sleep: (ms: number) => Promise<void>;
|
||||
|
||||
constructor(private readonly deps: StartupAdmissionDeps, readonly options: StartupAdmissionOptions = startupAdmissionOptions()) {
|
||||
this.now = deps.now ?? Date.now;
|
||||
this.sleep = deps.sleep ?? (ms => new Promise(resolve => setTimeout(resolve, ms)));
|
||||
for (const snapshot of deps.history()) {
|
||||
this.latest.set(snapshot.nodeId, snapshot);
|
||||
if (["starting", "uncertain"].includes(snapshot.phase) || (snapshot.phase === "settling" && snapshot.dispatchedAt !== undefined)) this.unresolved.set(snapshot.nodeId, snapshot);
|
||||
}
|
||||
}
|
||||
|
||||
owns(nodeId: string): boolean { return this.scope.getStore()?.has(nodeId) === true; }
|
||||
status(nodeId: string): StartupAdmissionSnapshot | null { return this.latest.get(nodeId) ?? null; }
|
||||
pending(nodeId: string): boolean { return this.entries.has(nodeId); }
|
||||
|
||||
/** Called synchronously at the native-input / incumbent-stop boundary, under the owning lease. */
|
||||
beforeDispatch(nodeId: string, nativeTarget?: string): void {
|
||||
const entry = this.entries.get(nodeId);
|
||||
if (!entry || !this.owns(nodeId)) return;
|
||||
if (entry.cancelled) throw new StartupAdmissionError("startup_cancelled", "Cancelled before native dispatch; no session stopped.");
|
||||
const targets = entry.snapshot.nativeTargets ?? (entry.snapshot.nativeTarget ? [entry.snapshot.nativeTarget] : []);
|
||||
if (entry.snapshot.dispatchedAt !== undefined && (!nativeTarget || targets.includes(nativeTarget))) return;
|
||||
entry.snapshot.dispatchedAt ??= this.now();
|
||||
entry.snapshot.nativeTarget = nativeTarget ?? entry.snapshot.nativeTarget;
|
||||
entry.snapshot.nativeTargets = nativeTarget ? [...targets, nativeTarget] : targets;
|
||||
this.record(entry, "starting", "Native launch or incumbent stop is being dispatched");
|
||||
}
|
||||
|
||||
run<T>(nodeId: string, kind: string, work: () => Promise<T>, requestKey = ""): Promise<T> {
|
||||
if (this.owns(nodeId)) return work();
|
||||
requestKey = createHash("sha256").update(requestKey).digest("hex");
|
||||
const selection = this.deps.selection(nodeId);
|
||||
const existing = this.entries.get(nodeId);
|
||||
if (existing) {
|
||||
// A different caller already holding the lease cannot join work waiting to acquire that lease.
|
||||
if (this.deps.ownsLifecycle?.(nodeId) || existing.snapshot.selection !== selection || existing.snapshot.kind !== kind || existing.snapshot.requestKey !== requestKey) {
|
||||
return Promise.reject(new StartupAdmissionError("startup_in_progress", "Another startup operation owns this seat; inspect its admission status."));
|
||||
}
|
||||
return existing.result as Promise<T>;
|
||||
}
|
||||
let resolve!: Entry["resolve"], reject!: Entry["reject"];
|
||||
const result = new Promise<unknown>((yes, no) => { resolve = yes; reject = no; });
|
||||
const entry: Entry = { snapshot: { attemptId: randomUUID(), nodeId, kind, selection, requestKey,
|
||||
phase: "queued", reason: "Waiting for startup admission", observedAt: this.now() }, work, resolve, reject, result, continueOne: false };
|
||||
this.entries.set(nodeId, entry);
|
||||
const enqueue = () => {
|
||||
if (entry.cancelled) throw new StartupAdmissionError("startup_cancelled", "Cancelled while waiting for the lifecycle lease; no native input sent.");
|
||||
entry.waitingForLifecycle = false;
|
||||
this.queue.push(entry);
|
||||
void this.pump();
|
||||
return result;
|
||||
};
|
||||
try {
|
||||
// Do not overwrite an interrupted native attempt with a seemingly harmless queued event.
|
||||
if (!this.unresolved.has(nodeId)) this.record(entry, "queued", this.active ? "Waiting behind another startup" : "Waiting for startup admission");
|
||||
if (this.deps.lifecycle) {
|
||||
entry.waitingForLifecycle = true;
|
||||
void this.deps.lifecycle(nodeId, enqueue).catch(error => {
|
||||
if (this.entries.get(nodeId) !== entry) return;
|
||||
this.entries.delete(nodeId);
|
||||
try { this.record(entry, entry.cancelled ? "cancelled" : "complete", "Startup stopped before acquiring admission; no native input sent"); } catch { /* Preserve the original lease/storage failure. */ }
|
||||
entry.reject(error);
|
||||
});
|
||||
} else enqueue();
|
||||
} catch (error) { this.entries.delete(nodeId); entry.reject(error); }
|
||||
return result as Promise<T>;
|
||||
}
|
||||
|
||||
control(nodeId: string, action: "cancel" | "continue"): boolean {
|
||||
const entry = this.entries.get(nodeId);
|
||||
if (!entry || (!this.queue.includes(entry) && !(action === "cancel" && (entry.waitingForLifecycle || entry.snapshot.phase === "preparing")))) return false;
|
||||
if (action === "cancel") {
|
||||
if (!this.queue.includes(entry)) {
|
||||
entry.cancelled = true;
|
||||
this.record(entry, "cancelled", "Cancelled before native dispatch; no session stopped");
|
||||
if (entry.waitingForLifecycle) entry.reject(new StartupAdmissionError("startup_cancelled", "Cancelled while waiting for the lifecycle lease; no native input sent."));
|
||||
return true;
|
||||
}
|
||||
this.queue.splice(this.queue.indexOf(entry), 1);
|
||||
this.entries.delete(nodeId);
|
||||
// Keep an interrupted attempt's custody even when its waiting retry is cancelled.
|
||||
if (!this.unresolved.has(nodeId)) this.record(entry, "cancelled", "Cancelled before native dispatch; no session stopped");
|
||||
entry.reject(new StartupAdmissionError("startup_cancelled", "Startup cancelled before dispatch; no native input sent."));
|
||||
} else {
|
||||
if (this.unresolved.has(nodeId)) return false;
|
||||
entry.continueOne = true;
|
||||
this.record(entry, "held", "Continue one selected: memory protection may be insufficient; waiting for the concurrency slot");
|
||||
}
|
||||
void this.pump();
|
||||
return true;
|
||||
}
|
||||
|
||||
private record(entry: Entry, phase: AdmissionPhase, reason: string, memory?: StartupMemory): void {
|
||||
const unchanged = entry.snapshot.phase === phase && entry.snapshot.reason === reason;
|
||||
entry.snapshot = { ...entry.snapshot, phase, reason, observedAt: this.now(), ...(memory ? { memory } : {}) };
|
||||
if (memory) entry.snapshot.protectionUnavailable = this.unavailable(memory);
|
||||
this.latest.set(entry.snapshot.nodeId, entry.snapshot);
|
||||
// Status reads retain each fresh sample; a prolonged hold need not append an event every second.
|
||||
if (phase === "held" && unchanged && this.now() - (this.lastPersistedAt.get(entry.snapshot.nodeId) ?? 0) < 5000) return;
|
||||
const seq = this.deps.persist(entry.snapshot);
|
||||
this.lastPersistedAt.set(entry.snapshot.nodeId, this.now());
|
||||
if (phase === "starting" && typeof seq === "number") entry.snapshot.dispatchSeq = seq;
|
||||
}
|
||||
|
||||
private async sample(): Promise<StartupMemory> {
|
||||
try { return await this.deps.observe(); }
|
||||
catch (error) { return { observedAt: this.now(), source: "unavailable", availableBytes: null, totalBytes: null,
|
||||
pressure: "unknown", error: error instanceof Error ? error.message : String(error) }; }
|
||||
}
|
||||
|
||||
private invalidHeadroom(sample: StartupMemory): boolean {
|
||||
return this.now() - sample.observedAt > this.options.staleMs || sample.observedAt > this.now()
|
||||
|| sample.availableBytes === null || sample.totalBytes === null || !Number.isFinite(sample.availableBytes)
|
||||
|| !Number.isFinite(sample.totalBytes) || sample.totalBytes <= 0 || sample.availableBytes < 0;
|
||||
}
|
||||
|
||||
private unavailable(sample: StartupMemory): boolean {
|
||||
return this.invalidHeadroom(sample) || sample.pressure === "unknown";
|
||||
}
|
||||
|
||||
private shortage(sample: StartupMemory): boolean {
|
||||
// A known pressure warning still holds even if headroom telemetry is incomplete.
|
||||
const fresh = this.now() - sample.observedAt <= this.options.staleMs && sample.observedAt <= this.now();
|
||||
return (fresh && sample.pressure === "high") || (!this.invalidHeadroom(sample)
|
||||
&& sample.availableBytes! < sample.totalBytes! * this.options.minimumAvailablePercent / 100);
|
||||
}
|
||||
|
||||
private retryLater(): void {
|
||||
if (this.timer) return;
|
||||
this.timer = setTimeout(() => { this.timer = undefined; void this.pump(); }, 1000);
|
||||
this.timer.unref?.();
|
||||
}
|
||||
|
||||
private async pump(): Promise<void> {
|
||||
if (this.pumping) return;
|
||||
this.pumping = true;
|
||||
try {
|
||||
while (this.queue.length && this.active < this.options.concurrency) {
|
||||
const entry = this.queue[0]!;
|
||||
for (const [nodeId, snapshot] of this.unresolved) {
|
||||
const state = await this.deps.reconcile(snapshot).catch(() => "unknown" as const);
|
||||
if (state === "absent" || state === "ready") {
|
||||
this.unresolved.delete(nodeId);
|
||||
if (state === "ready" && nodeId === entry.snapshot.nodeId) {
|
||||
this.queue.shift(); this.entries.delete(nodeId);
|
||||
entry.reject(new StartupAdmissionError("startup_already_observed", "The earlier launch is now ready. No second launch was sent; inspect the seat."));
|
||||
this.retryLater();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!this.entries.has(entry.snapshot.nodeId)) continue;
|
||||
// Even continue-one cannot authorize a duplicate of a live/unknown interrupted attempt.
|
||||
if (this.unresolved.has(entry.snapshot.nodeId)) {
|
||||
this.queue.shift(); this.entries.delete(entry.snapshot.nodeId);
|
||||
entry.reject(new StartupAdmissionError("startup_unreconciled", "An earlier native launch is live or unknown. Reconcile its original session before retrying; no second launch was sent."));
|
||||
continue;
|
||||
}
|
||||
const memory = await this.sample();
|
||||
// A cancellation may arrive during an OS read.
|
||||
if (!this.queue.includes(entry)) continue;
|
||||
const unavailable = this.unavailable(memory);
|
||||
entry.snapshot.protectionUnavailable = unavailable;
|
||||
if (!entry.continueOne && (this.unresolved.size || this.shortage(memory) || (unavailable && this.options.strictTelemetry))) {
|
||||
const reason = this.unresolved.size ? "An earlier launch is unresolved"
|
||||
: unavailable && this.options.strictTelemetry ? "Memory telemetry unavailable; strict telemetry wait selected"
|
||||
: this.active ? "Memory headroom held behind another startup" : "One-seat memory deficit: no other startup holds a slot";
|
||||
this.record(entry, "held", `${reason}. Cancel before dispatch or deliberately continue one.`, memory);
|
||||
this.retryLater(); break;
|
||||
}
|
||||
// Missing observations always serialize, even when healthy concurrency was configured higher.
|
||||
if (unavailable && this.active) { this.record(entry, "held", "Memory protection unavailable: serial readiness and pacing only", memory); break; }
|
||||
if (this.deps.selection(entry.snapshot.nodeId) !== entry.snapshot.selection) {
|
||||
this.queue.shift(); this.entries.delete(entry.snapshot.nodeId);
|
||||
this.record(entry, "cancelled", "Seat selection changed before dispatch; refresh rather than start another occupant");
|
||||
entry.reject(new StartupAdmissionError("startup_selection_changed", "The seat selection changed while waiting; refresh before starting it."));
|
||||
continue;
|
||||
}
|
||||
this.record(entry, "preparing", unavailable ? "Memory protection unavailable: serialized startup with bounded pacing"
|
||||
: entry.continueOne ? "Continuing one with the observed memory risk" : "Startup admitted", memory);
|
||||
this.queue.shift(); this.active++;
|
||||
void this.execute(entry);
|
||||
}
|
||||
} catch (error) {
|
||||
const entry = this.queue.shift();
|
||||
if (entry) { this.entries.delete(entry.snapshot.nodeId); entry.reject(error); }
|
||||
if (this.queue.length) this.retryLater();
|
||||
} finally { this.pumping = false; }
|
||||
}
|
||||
|
||||
private async execute(entry: Entry): Promise<void> {
|
||||
try {
|
||||
let result = await this.scope.run(new Set([entry.snapshot.nodeId]), entry.work);
|
||||
if (entry.cancelled && entry.snapshot.dispatchedAt === undefined) throw new StartupAdmissionError("startup_cancelled", "Cancelled before native dispatch; no session stopped.");
|
||||
this.record(entry, "settling", "Native operation returned; readiness alone does not prove memory settling");
|
||||
const started = this.now();
|
||||
// Bounded pacing also applies with missing telemetry. No fake claim of a measured peak end.
|
||||
await this.sleep(Math.min(this.options.settleMs, this.options.settleLimitMs));
|
||||
let memory = await this.sample();
|
||||
while (this.shortage(memory) && this.now() - started < this.options.settleLimitMs) {
|
||||
this.record(entry, "settling", "Waiting for observed memory recovery after native readiness", memory);
|
||||
await this.sleep(Math.min(1000, Math.max(0, this.options.settleLimitMs - (this.now() - started))));
|
||||
memory = await this.sample();
|
||||
}
|
||||
const failed = typeof result === "object" && result !== null
|
||||
&& ((result as { ok?: boolean }).ok === false || ["failed", "attention_required", "awaiting-decision"].includes((result as { status?: string }).status ?? ""));
|
||||
if (failed && entry.snapshot.dispatchedAt !== undefined) {
|
||||
const state = await this.deps.reconcile(entry.snapshot).catch(() => "unknown" as const);
|
||||
if (state === "live" || state === "unknown") {
|
||||
this.record(entry, "uncertain", "Native startup did not confirm success; reconcile the live/unknown attempt before retrying", memory);
|
||||
this.unresolved.set(entry.snapshot.nodeId, entry.snapshot);
|
||||
} else this.record(entry, "complete", "Startup returned a refusal; no unresolved native process observed", memory);
|
||||
} else this.record(entry, "complete", "Startup finished; bounded pacing completed, not a guarantee against a later memory peak", memory);
|
||||
if (entry.snapshot.protectionUnavailable && result && typeof result === "object" && !Array.isArray(result)) {
|
||||
const previous = (result as { warnings?: unknown }).warnings;
|
||||
const warnings = Array.isArray(previous) ? previous.filter((value): value is string => typeof value === "string") : [];
|
||||
result = { ...result, warnings: [...new Set([...warnings, "Startup memory protection unavailable; this daemon serialized starts and used bounded pacing. Inspect startup admission status before starting more seats."])] };
|
||||
}
|
||||
entry.resolve(result);
|
||||
} catch (error) {
|
||||
const dispatched = entry.snapshot.dispatchedAt !== undefined;
|
||||
try { this.record(entry, dispatched ? "uncertain" : entry.cancelled ? "cancelled" : "complete",
|
||||
dispatched ? "Startup outcome is unknown; reconcile before retrying" : "Startup stopped before native dispatch; no native input sent"); } catch { /* Preserve the original failure if event storage itself failed. */ }
|
||||
if (dispatched) this.unresolved.set(entry.snapshot.nodeId, entry.snapshot);
|
||||
entry.reject(error);
|
||||
} finally {
|
||||
this.active--; this.entries.delete(entry.snapshot.nodeId);
|
||||
void this.pump();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import os from "node:os";
|
||||
|
||||
export interface StartupMemory {
|
||||
observedAt: number;
|
||||
source: string;
|
||||
availableBytes: number | null;
|
||||
totalBytes: number | null;
|
||||
pressure: "normal" | "high" | "unknown";
|
||||
error?: string;
|
||||
}
|
||||
|
||||
const exec = promisify(execFile);
|
||||
|
||||
export function linuxStartupMemory(text: string, now: number): StartupMemory {
|
||||
const field = (key: string) => {
|
||||
const match = text.match(new RegExp(`^${key}:\\s+(\\d+) kB$`, "m"));
|
||||
return match ? Number(match[1]) * 1024 : null;
|
||||
};
|
||||
return { observedAt: now, source: "linux:MemAvailable", availableBytes: field("MemAvailable"),
|
||||
totalBytes: field("MemTotal"), pressure: "unknown" };
|
||||
}
|
||||
|
||||
export function macStartupMemory(text: string, pressure: string, total: number, now: number): StartupMemory {
|
||||
// Use the system-wide percentage reported by -Q, not vm_stat's free-page count.
|
||||
const match = text.match(/System-wide memory free percentage:\s*(\d+)%/);
|
||||
const percent = match ? Number(match[1]) : NaN;
|
||||
return { observedAt: now, source: "macos:memory_pressure-Q", totalBytes: total,
|
||||
availableBytes: Number.isFinite(percent) && percent >= 0 && percent <= 100 ? total * percent / 100 : null,
|
||||
pressure: pressure.trim() === "1" ? "normal" : ["2", "4"].includes(pressure.trim()) ? "high" : "unknown" };
|
||||
}
|
||||
|
||||
/** Bounded read-only OS observations; never invoke memory_pressure's allocation/simulation options. */
|
||||
export async function observeStartupMemory(): Promise<StartupMemory> {
|
||||
try {
|
||||
if (process.platform === "linux") {
|
||||
const sample = linuxStartupMemory(await readFile("/proc/meminfo", "utf8"), Date.now());
|
||||
try {
|
||||
const psi = await readFile("/proc/pressure/memory", "utf8");
|
||||
const some = psi.match(/^some avg10=([\d.]+)/m);
|
||||
const full = psi.match(/^full avg10=([\d.]+)/m);
|
||||
if (some && full) {
|
||||
sample.pressure = Number(some[1]) >= 10 || Number(full[1]) >= 1 ? "high" : "normal";
|
||||
sample.source += "+psi";
|
||||
}
|
||||
} catch { /* No PSI means pressure protection is unavailable; admission degrades visibly. */ }
|
||||
// A constrained daemon cannot spend memory outside its own cgroup ceiling.
|
||||
try {
|
||||
const group = (await readFile("/proc/self/cgroup", "utf8")).match(/^0::(.+)$/m)?.[1];
|
||||
if (group && !group.split("/").includes("..")) {
|
||||
const base = `/sys/fs/cgroup${group === "/" ? "" : group}`;
|
||||
const [max, current] = await Promise.all([readFile(`${base}/memory.max`, "utf8"), readFile(`${base}/memory.current`, "utf8")]);
|
||||
if (/^\d+\s*$/.test(max) && /^\d+\s*$/.test(current)) {
|
||||
const remaining = Math.max(0, Number(max) - Number(current));
|
||||
sample.totalBytes = sample.totalBytes === null ? Number(max) : Math.min(sample.totalBytes, Number(max));
|
||||
sample.availableBytes = sample.availableBytes === null ? remaining : Math.min(sample.availableBytes, remaining);
|
||||
sample.source += "+cgroup";
|
||||
}
|
||||
}
|
||||
} catch { /* Host observation remains usable when this kernel has no readable cgroup v2 ceiling. */ }
|
||||
return sample;
|
||||
}
|
||||
if (process.platform === "darwin") {
|
||||
const [available, pressure] = await Promise.all([
|
||||
exec("/usr/bin/memory_pressure", ["-Q"], { timeout: 1500, maxBuffer: 8192 }),
|
||||
exec("/usr/sbin/sysctl", ["-n", "kern.memorystatus_vm_pressure_level"], { timeout: 1500, maxBuffer: 1024 }),
|
||||
]);
|
||||
return macStartupMemory(available.stdout, pressure.stdout, os.totalmem(), Date.now());
|
||||
}
|
||||
throw new Error(`No startup memory observer for ${process.platform}`);
|
||||
} catch (error) {
|
||||
return { observedAt: Date.now(), source: process.platform, availableBytes: null, totalBytes: null,
|
||||
pressure: "unknown", error: error instanceof Error ? error.message : String(error) };
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { StartupAdmissionError } from "./startup-admission.js";
|
||||
import { nonInterruptiveNotice } from "../adapters/non-interruptive.js";
|
||||
import { planModelAliasNotice } from "../adapters/yolo-mode.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
@@ -333,8 +334,17 @@ export class StartupOrchestrator {
|
||||
|
||||
async startNode(input: StartupInput): Promise<StartupResult> {
|
||||
const warnings: string[] = [];
|
||||
const result = await this.startNodeWithWarnings(input, warnings);
|
||||
return { ...result, ...(warnings.length ? { warnings: [...new Set(warnings)] } : {}) };
|
||||
const work = () => this.startNodeWithWarnings(input, warnings);
|
||||
let result: StartupResult;
|
||||
try {
|
||||
result = this.tmuxAdapter.startupAdmission
|
||||
? await this.tmuxAdapter.startupAdmission.run(input.nodeId, "startup", work, JSON.stringify([input.sessionId, input.binding.launchGeneration]))
|
||||
: await work();
|
||||
} catch (error) {
|
||||
if (!(error instanceof StartupAdmissionError)) throw error;
|
||||
return { ok: false, startupStatus: "attention_required", errors: [error.message] };
|
||||
}
|
||||
return { ...result, ...(warnings.length ? { warnings: [...new Set([...(result.warnings ?? []), ...warnings])] } : {}) };
|
||||
}
|
||||
|
||||
private async startNodeWithWarnings(input: StartupInput, warnings: string[]): Promise<StartupResult> {
|
||||
@@ -607,6 +617,7 @@ export class StartupOrchestrator {
|
||||
if (input.beforeNativeLaunch && !input.beforeNativeLaunch()) {
|
||||
return this.fail(input, "attention_required", ["The selected seat, settings or history changed after preparation; native resume was not dispatched."], undefined, false, undefined, false, "harness_launch", true);
|
||||
}
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(input.nodeId, input.binding.tmuxSession ?? undefined);
|
||||
const launchResult = await input.adapter.launchHarness(input.binding, {
|
||||
name: input.sessionName ?? input.binding.tmuxSession ?? "",
|
||||
...(launchPrompt ? { firstInputText: launchPrompt.prompt } : {}),
|
||||
|
||||
@@ -191,6 +191,13 @@ export class SuccessorSessionLauncher {
|
||||
* post-fork token, never the parent's. Absent → a plain fresh launch. */
|
||||
forkSource?: ForkSource;
|
||||
}): Promise<SuccessorLaunchResult> {
|
||||
const admission = this.tmuxAdapter.startupAdmission;
|
||||
if (admission && !admission.owns(input.node.id)) {
|
||||
try { return await admission.run(input.node.id, "successor", () => this.createSuccessor(input), JSON.stringify([input.departingSessionName, input.occupantGeneration, input.forkSource])); }
|
||||
catch (error) { return { ok: false, code: "startup_admission", message: (error as Error).message, step: "create_successor", replacementStarted: false }; }
|
||||
}
|
||||
const guard = this.tmuxAdapter.deliveryGuard;
|
||||
if (guard && !guard.ownsLifecycle(input.node.id)) return guard.lifecycle([input.node.id], () => this.createSuccessor(input));
|
||||
// CUTOVER MODEL (plan 411c43de): a SEAT = one durable tmux session; the successor takes over the
|
||||
// retiree's EXACT pane via respawn-pane, so the canonical session name is PRESERVED (no -h shuffle)
|
||||
// and native scrollback survives — predecessor history stays above the successor boot (the money
|
||||
@@ -245,6 +252,7 @@ export class SuccessorSessionLauncher {
|
||||
// money-proof). The swap: set remain-on-exit so the pane survives the exit → graceful SIGTERM
|
||||
// ("exits in place"; also protects the provider session state the successor's resume depends on) →
|
||||
// bounded-timeout SIGKILL fallback (pinned degraded path) → respawn-pane (no -k) on the dead pane.
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(input.node.id, input.departingSessionName);
|
||||
const terminated = await this.terminateRetiree(pane.id);
|
||||
if (!terminated.ok) {
|
||||
return { ok: false, code: "retiree_not_terminated", step: "create_successor", message: terminated.message, replacementStarted: false };
|
||||
@@ -335,6 +343,13 @@ export class SuccessorSessionLauncher {
|
||||
staged: Omit<StagedLaunch, "canonicalSession">;
|
||||
}): Promise<{ ok: true; tmuxPane: string; resumeToken?: string; resumeType?: string; appliedLaunch?: AppliedLaunchObservation; notes?: string[] }
|
||||
| { ok: false; code: string; message: string }> {
|
||||
const admission = this.tmuxAdapter.startupAdmission;
|
||||
if (admission && !admission.owns(input.node.id)) {
|
||||
try { return await admission.run(input.node.id, "staged successor", () => this.stageSuccessor(input), input.occupantGeneration); }
|
||||
catch (error) { return { ok: false, code: "startup_admission", message: (error as Error).message }; }
|
||||
}
|
||||
const guard = this.tmuxAdapter.deliveryGuard;
|
||||
if (guard && !guard.ownsLifecycle(input.node.id)) return guard.lifecycle([input.node.id], () => this.stageSuccessor(input));
|
||||
const env = compactEnv({
|
||||
OPENRIG_NODE_ID: input.node.id,
|
||||
OPENRIG_SESSION_NAME: input.canonicalSession,
|
||||
@@ -443,6 +458,7 @@ export class SuccessorSessionLauncher {
|
||||
|
||||
let launch: Awaited<ReturnType<RuntimeAdapter["launchHarness"]>>;
|
||||
try {
|
||||
this.tmuxAdapter.startupAdmission?.beforeDispatch(node.id, tmuxSession);
|
||||
launch = await adapter.launchHarness(binding, staged
|
||||
? { name: staged.canonicalSession }
|
||||
: { name: tmuxSession, ...(forkSource ? { forkSource } : { freshInPlace: true }) });
|
||||
|
||||
@@ -254,6 +254,7 @@ export type RigEvent =
|
||||
// AgentSpec reboot events — pods + startup + continuity
|
||||
| { type: "pod.created"; rigId: string; podId: string; namespace: string; label: string }
|
||||
| { type: "pod.deleted"; rigId: string; podId: string }
|
||||
| { type: "node.startup_admission"; rigId: string; nodeId: string; admission: import("./startup-admission.js").StartupAdmissionSnapshot }
|
||||
| { type: "node.startup_pending"; rigId: string; nodeId: string; startupProof?: StartupProofSelection }
|
||||
| { type: "node.startup_first_input_passed"; rigId: string; nodeId: string; sessionId: string; path: string; submission: "unconfirmed" }
|
||||
| { type: "node.startup_model_passed"; rigId: string; nodeId: string; sessionId: string; model: string; resume: boolean }
|
||||
|
||||
@@ -178,6 +178,7 @@ nodesRoutes.get("/", observeInventoryHandler, async (c) => {
|
||||
const modelReader = c.get("currentModelReader" as never) as CurrentModelReader | undefined;
|
||||
const observations = modelReader ? await modelReader.observeInventory(withTerminalAndWork) : null;
|
||||
const nodes = withTerminalAndWork.map((n, i) => ({ ...n, hostSelfId,
|
||||
startupAdmission: deps.tmuxAdapter.startupAdmission?.status(n.nodeId) ?? null,
|
||||
modelObservation: observations?.[i] ?? unknownModelObservation("observer-unavailable"),
|
||||
}));
|
||||
return c.json(nodes);
|
||||
@@ -204,6 +205,7 @@ nodesRoutes.get("/:logicalId", async (c) => {
|
||||
agentActivity: deps.agentActivityStore,
|
||||
});
|
||||
Object.assign(detail, {
|
||||
startupAdmission: deps.tmuxAdapter.startupAdmission?.status(detail.nodeId) ?? null,
|
||||
needsAttention: detailWithTerminalAndWork?.needsAttention,
|
||||
arbitratedNeedsInput: detailWithTerminalAndWork?.arbitratedNeedsInput,
|
||||
needsInputEvidence: detailWithTerminalAndWork?.needsInputEvidence,
|
||||
@@ -260,6 +262,23 @@ nodesRoutes.get("/:logicalId", async (c) => {
|
||||
return c.json(detail);
|
||||
});
|
||||
|
||||
// Startup admission controls never acquire the seat lifecycle lease held by a waiting launch.
|
||||
nodesRoutes.post("/:logicalId/startup-admission", async c => {
|
||||
const { rigRepo, tmuxAdapter } = getDeps(c);
|
||||
const rigId = c.req.param("rigId")!;
|
||||
const ref = c.req.param("logicalId")!;
|
||||
const node = rigRepo.getRig(rigId)?.nodes.find(node => node.logicalId === ref || node.id === ref);
|
||||
if (!node) return c.json({ error: "Seat not found" }, 404);
|
||||
const admission = tmuxAdapter.startupAdmission;
|
||||
if (!admission) return c.json({ error: "Startup admission is unavailable on this daemon" }, 503);
|
||||
const body = await c.req.json().catch(() => ({})) as { action?: string };
|
||||
if (!["status", "cancel", "continue"].includes(body.action ?? "")) return c.json({ error: "Use status, cancel or continue" }, 400);
|
||||
if (body.action !== "status" && !admission.control(node.id, body.action as "cancel" | "continue")) {
|
||||
return c.json({ error: "No waiting request is eligible for this action. Native input is never cancelled or replayed here.", admission: admission.status(node.id) }, 409);
|
||||
}
|
||||
return c.json({ ok: true, admission: admission.status(node.id), requestActive: admission.pending(node.id), options: admission.options });
|
||||
});
|
||||
|
||||
// POST /api/rigs/:rigId/nodes/:logicalId/launch
|
||||
nodesRoutes.post("/:logicalId/launch", async (c) => {
|
||||
const rigId = c.req.param("rigId")!;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { daemonStartupAdmission } from "./domain/daemon-startup-admission.js";
|
||||
import { EnvironmentStartup } from "./domain/environment-startup.js";
|
||||
import { EnvironmentPlatform } from "./domain/environment-platform.js";
|
||||
import { OpenShellDriver } from "./adapters/openshell-driver.js";
|
||||
@@ -557,6 +558,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
|
||||
const deliveryGuard = new SeatDeliveryGuard(db, target => resolveGuardTarget(db, target));
|
||||
deliveryGuard.recoverActivation();
|
||||
tmuxAdapter.deliveryGuard = deliveryGuard;
|
||||
tmuxAdapter.startupAdmission = daemonStartupAdmission(db, eventBus, sessionRegistry, tmuxAdapter);
|
||||
|
||||
// Slice 15 — Seat-activity service for the `terminal-active` primitive.
|
||||
// Lives at module scope so the projection chain (PsProjectionService,
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { createFullTestDb } from "./helpers/test-app.js";
|
||||
import { EventBus } from "../src/domain/event-bus.js";
|
||||
import { RigRepository } from "../src/domain/rig-repository.js";
|
||||
import { SessionRegistry } from "../src/domain/session-registry.js";
|
||||
import { daemonStartupAdmission } from "../src/domain/daemon-startup-admission.js";
|
||||
import type { StartupAdmissionSnapshot } from "../src/domain/startup-admission.js";
|
||||
import type { TmuxAdapter } from "../src/adapters/tmux.js";
|
||||
|
||||
vi.mock("../src/domain/startup-memory.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../src/domain/startup-memory.js")>(),
|
||||
observeStartupMemory: async () => ({ observedAt: Date.now(), source: "fixture", totalBytes: 1000, availableBytes: 800, pressure: "normal" }),
|
||||
}));
|
||||
vi.mock("../src/domain/native-process-lineage.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../src/domain/native-process-lineage.js")>(), verifyPaneIdleShell: async () => false,
|
||||
}));
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
function fixture() {
|
||||
vi.stubEnv("OPENRIG_STARTUP_SETTLE_MS", "0"); vi.stubEnv("OPENRIG_STARTUP_SETTLE_LIMIT_MS", "0");
|
||||
const db = createFullTestDb(), events = new EventBus(db), registry = new SessionRegistry(db), rigs = new RigRepository(db);
|
||||
const rig = rigs.createRig("admission-fixture"), node = rigs.addNode(rig.id, "worker", { runtime: "codex", model: "selected-model", effort: "high" });
|
||||
const probeSession = vi.fn(async (_target: string) => ({ state: "absent" as "present" | "absent" | "unknown" }));
|
||||
const tmux = { probeSession } as unknown as TmuxAdapter;
|
||||
return { db, events, registry, rig, node, tmux, probeSession, controller: () => daemonStartupAdmission(db, events, registry, tmux) };
|
||||
}
|
||||
it("persists native dispatch and completion on the existing event surface with the original node settings", async () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
const controller = f.controller();
|
||||
expect(await controller.run(f.node.id, "startup", async () => { controller.beforeDispatch(f.node.id, "worker@admission-fixture"); return "original conversation"; })).toBe("original conversation");
|
||||
const rows = f.db.prepare("SELECT payload FROM events WHERE type='node.startup_admission' ORDER BY seq").all() as Array<{payload: string}>;
|
||||
const snapshots = rows.map(row => JSON.parse(row.payload).admission as StartupAdmissionSnapshot);
|
||||
expect(snapshots.some(row => row.phase === "starting" && row.nativeTarget === "worker@admission-fixture")).toBe(true);
|
||||
expect(snapshots.at(-1)?.phase).toBe("complete");
|
||||
expect(f.controller().status(f.node.id)?.phase).toBe("complete");
|
||||
expect(f.db.prepare("SELECT model,effort FROM nodes WHERE id=?").get(f.node.id)).toEqual({ model: "selected-model", effort: "high" });
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
it.each(["live", "unknown"])("retains interrupted custody when an old target is gone but the later target is %s", async state => {
|
||||
const f = fixture();
|
||||
try {
|
||||
const admission: StartupAdmissionSnapshot = { attemptId: "interrupted", nodeId: f.node.id, kind: "restore", selection: "original", phase: "starting", reason: "lost reply", observedAt: Date.now(), dispatchedAt: Date.now(), nativeTarget: "new-target", nativeTargets: ["old-target", "new-target"] };
|
||||
f.events.emit({ type: "node.startup_admission", rigId: f.rig.id, nodeId: f.node.id, admission });
|
||||
f.probeSession.mockImplementation(async target => ({ state: target === "old-target" ? "absent" : state === "live" ? "present" : "unknown" }));
|
||||
const controller = f.controller(), work = vi.fn(async () => "must not repeat");
|
||||
await expect(controller.run(f.node.id, "restore", work)).rejects.toMatchObject({ code: "startup_unreconciled" });
|
||||
expect(work).not.toHaveBeenCalled(); expect(controller.status(f.node.id)?.attemptId).toBe("interrupted");
|
||||
expect(f.probeSession.mock.calls.map(call => call[0])).toEqual(["old-target", "new-target"]);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
@@ -383,6 +383,14 @@ describe("SeatLifecycleService.launchFresh", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("never enters startup admission or launches from a fresh preview", async () => {
|
||||
const seat = seedSeat({ clean: true });
|
||||
const run = vi.fn(async () => { throw new Error("a preview cannot admit native work"); });
|
||||
Object.assign(tmux, { startupAdmission: { owns: () => false, run } });
|
||||
const result = await service.planFresh({ seatRef: seat.sessionName, fresh: true });
|
||||
expect(result).toMatchObject({ ok: true, planOnly: true });
|
||||
expect(run).not.toHaveBeenCalled(); expect(tmux.createSession).not.toHaveBeenCalled();
|
||||
});
|
||||
it.each([false, true])("previews fresh launch without changing target, siblings or startup state (live=%s)", async live => {
|
||||
const seat = seedSeat({ clean: !live });
|
||||
const sibling = rigRepo.addNode(seat.rig.id, "dev.qa", { runtime: "claude-code", cwd: "/sibling" });
|
||||
|
||||
@@ -273,6 +273,18 @@ function state(activity: string, needs = 0): ArbitratedSeatState {
|
||||
}
|
||||
|
||||
describe("rig seat restart --keep-conversation: gates before any signal", () => {
|
||||
it("obtains shared admission before stopping a healthy in-place restart", async () => {
|
||||
const w = world();
|
||||
const run = vi.fn(async () => { throw new Error("Cancelled before native dispatch"); });
|
||||
Object.assign(w.tmux, { startupAdmission: { owns: () => false, run } });
|
||||
const result = await w.service.restartInPlace({ seatRef: SEAT, reason: "selected restart" });
|
||||
expect(result).toMatchObject({ ok: false, code: "startup_admission" });
|
||||
expect(run).toHaveBeenCalledTimes(1);
|
||||
expect(w.relaunched).toBe(false);
|
||||
expect(w.launches).toEqual([]);
|
||||
expect(w.events()).toEqual([]);
|
||||
});
|
||||
|
||||
it("plan reports activity, child processes and settings, and changes nothing", async () => {
|
||||
const w = world({ activity: state("working") });
|
||||
w.setRows([shellRow(100, 200), codexRow(200, 100, `-m gpt-5.5 -s workspace-write resume ${TOKEN}`),
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { StartupAdmission, startupAdmissionOptions, type StartupAdmissionSnapshot, type StartupAdmissionDeps } from "../src/domain/startup-admission.js";
|
||||
import { linuxStartupMemory, macStartupMemory, type StartupMemory } from "../src/domain/startup-memory.js";
|
||||
|
||||
function deferred<T = void>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
const promise = new Promise<T>(yes => { resolve = yes; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
async function flush() { for (let i = 0; i < 20; i++) await Promise.resolve(); }
|
||||
|
||||
function harness(overrides: Partial<StartupAdmissionDeps> = {}, options = {}) {
|
||||
let now = 10000;
|
||||
const records: StartupAdmissionSnapshot[] = [];
|
||||
const healthy = (): StartupMemory => ({ observedAt: now, source: "fixture", totalBytes: 1000, availableBytes: 800, pressure: "normal" });
|
||||
const controller = new StartupAdmission({ observe: async () => healthy(), selection: () => "original-generation",
|
||||
persist: snapshot => { records.push(structuredClone(snapshot)); }, history: () => [], reconcile: async () => "absent",
|
||||
now: () => now, sleep: async ms => { now += ms; }, ...overrides },
|
||||
{ ...startupAdmissionOptions({}), settleMs: 10, settleLimitMs: 30, ...options });
|
||||
return { controller, records, healthy };
|
||||
}
|
||||
|
||||
describe("owning-daemon native startup admission", () => {
|
||||
it("deduplicates before waiting for an existing lifecycle lease without consuming a startup slot", async () => {
|
||||
const lease = deferred(); let leases = 0;
|
||||
const { controller } = harness({ lifecycle: async (_node, work) => { leases++; await lease.promise; return work(); } });
|
||||
let launches = 0;
|
||||
const first = controller.run("seat", "restore", async () => { launches++; return "original identity"; }, "same request");
|
||||
const duplicate = controller.run("seat", "restore", async () => { launches++; return "duplicate"; }, "same request");
|
||||
expect(duplicate).toBe(first); expect(leases).toBe(1); await flush(); expect(launches).toBe(0);
|
||||
lease.resolve(); expect(await duplicate).toBe("original identity"); expect(launches).toBe(1);
|
||||
});
|
||||
it("does not let a caller holding the lifecycle lease deadlock on a request waiting for that lease", async () => {
|
||||
const lease = deferred(); let owned = false;
|
||||
const { controller } = harness({ ownsLifecycle: () => owned, lifecycle: async (_node, work) => { await lease.promise; return work(); } });
|
||||
const pending = controller.run("seat", "restore", async () => "original");
|
||||
owned = true;
|
||||
await expect(controller.run("seat", "restore", async () => "second")).rejects.toMatchObject({ code: "startup_in_progress" });
|
||||
owned = false; lease.resolve(); expect(await pending).toBe("original");
|
||||
});
|
||||
it("cancels while waiting for a lifecycle lease before any launch or incumbent stop", async () => {
|
||||
const lease = deferred(); let effects = 0;
|
||||
const { controller } = harness({ lifecycle: async (_node, work) => { await lease.promise; return work(); } });
|
||||
const result = controller.run("seat", "restore", async () => { effects++; }).catch(error => error);
|
||||
expect(controller.control("seat", "cancel")).toBe(true);
|
||||
expect((await result).code).toBe("startup_cancelled");
|
||||
lease.resolve(); await flush(); expect(effects).toBe(0); expect(controller.pending("seat")).toBe(false);
|
||||
});
|
||||
it("refuses a different selected request instead of joining its result", async () => {
|
||||
const ready = deferred();
|
||||
const { controller } = harness();
|
||||
const first = controller.run("seat", "restore", async () => { await ready.promise; return "original"; }, "history-one");
|
||||
await expect(controller.run("seat", "restore", async () => "changed", "history-two")).rejects.toMatchObject({ code: "startup_in_progress" });
|
||||
ready.resolve(); expect(await first).toBe("original");
|
||||
});
|
||||
it("retains every dispatched target when a restore stops an old pane then launches another", async () => {
|
||||
const { controller, records } = harness();
|
||||
await controller.run("seat", "restore", async () => { controller.beforeDispatch("seat", "old-pane"); controller.beforeDispatch("seat", "new-pane"); });
|
||||
expect(records.at(-1)?.nativeTargets).toEqual(["old-pane", "new-pane"]);
|
||||
});
|
||||
it("serializes two rigs through one readiness and settling slot", async () => {
|
||||
const { controller, records } = harness();
|
||||
const ready = deferred();
|
||||
const calls: string[] = [];
|
||||
const a = controller.run("rig-a/seat", "restore", async () => { calls.push("a"); controller.beforeDispatch("rig-a/seat"); await ready.promise; return { ok: true }; });
|
||||
const b = controller.run("rig-b/seat", "successor", async () => { calls.push("b"); controller.beforeDispatch("rig-b/seat"); return { ok: true }; });
|
||||
await flush(); expect(calls).toEqual(["a"]);
|
||||
ready.resolve(); await Promise.all([a, b]); expect(calls).toEqual(["a", "b"]);
|
||||
const completeA = records.findIndex(r => r.nodeId === "rig-a/seat" && r.phase === "complete");
|
||||
const startB = records.findIndex(r => r.nodeId === "rig-b/seat" && r.phase === "starting");
|
||||
expect(startB).toBeGreaterThan(completeA);
|
||||
});
|
||||
|
||||
it("keeps the slot after readiness while fresh memory remains low, up to the settling bound", async () => {
|
||||
let reads = 0;
|
||||
const { controller, records } = harness({ observe: async () => ({ observedAt: 10000 + Math.max(0, reads - 1) * 10,
|
||||
source: "fixture", totalBytes: 1000, availableBytes: ++reads === 1 || reads > 3 ? 800 : 100, pressure: "normal" }) });
|
||||
await controller.run("seat", "startup", async () => ({ ok: true }));
|
||||
expect(records.some(r => r.phase === "settling" && r.reason.includes("memory recovery"))).toBe(true);
|
||||
expect(records.at(-1)?.phase).toBe("complete");
|
||||
});
|
||||
|
||||
it("surfaces a one-seat deficit before any dispatch and allows deliberate continue-one", async () => {
|
||||
const { controller } = harness({ observe: async () => ({ observedAt: 10000, source: "fixture", totalBytes: 1000, availableBytes: 100, pressure: "high" }) });
|
||||
let calls = 0;
|
||||
const attempt = controller.run("seat", "startup", async () => { calls++; return { ok: true }; });
|
||||
await flush(); expect(calls).toBe(0);
|
||||
expect(controller.status("seat")?.reason).toContain("One-seat memory deficit");
|
||||
expect(controller.control("seat", "continue")).toBe(true);
|
||||
await attempt; expect(calls).toBe(1);
|
||||
});
|
||||
|
||||
it("degraded telemetry serializes even with a larger configured cap and never permanently refuses", async () => {
|
||||
const { controller, records } = harness({ observe: async () => { throw new Error("no telemetry"); } }, { concurrency: 3 });
|
||||
const ready = deferred(); const calls: string[] = [];
|
||||
const a = controller.run("a", "startup", async () => { calls.push("a"); await ready.promise; return { ok: true }; });
|
||||
const b = controller.run("b", "startup", async () => { calls.push("b"); return { ok: true }; });
|
||||
await flush(); expect(calls).toEqual(["a"]);
|
||||
expect(records.some(r => r.protectionUnavailable && r.reason.includes("unavailable"))).toBe(true);
|
||||
ready.resolve(); await Promise.all([a, b]); expect(calls).toEqual(["a", "b"]);
|
||||
});
|
||||
|
||||
it("returns the protection-unavailable notice with ordinary startup results", async () => {
|
||||
const { controller } = harness({ observe: async () => { throw new Error("missing"); } });
|
||||
const result = await controller.run("seat", "startup", async () => ({ ok: true, warnings: ["existing warning"] }));
|
||||
expect(result.warnings).toContain("existing warning");
|
||||
expect(result.warnings.some(value => value.includes("memory protection unavailable"))).toBe(true);
|
||||
});
|
||||
it("caps even the minimum pacing at the configured settling bound", async () => {
|
||||
const sleeps: number[] = [];
|
||||
const { controller } = harness({ sleep: async ms => { sleeps.push(ms); } }, { settleMs: 100, settleLimitMs: 10 });
|
||||
await controller.run("seat", "startup", async () => ({ ok: true }));
|
||||
expect(sleeps).toEqual([10]);
|
||||
});
|
||||
it("a stale pressure reading degrades instead of becoming an indefinite pressure refusal", async () => {
|
||||
const { controller } = harness({ observe: async () => ({ observedAt: 0, source: "stale", totalBytes: 1000, availableBytes: 0, pressure: "high" }) });
|
||||
await expect(controller.run("seat", "startup", async () => "kept history")).resolves.toBe("kept history");
|
||||
expect(controller.status("seat")?.protectionUnavailable).toBe(true);
|
||||
});
|
||||
|
||||
it("strict telemetry waiting is opt-in and can be cancelled before dispatch", async () => {
|
||||
const { controller } = harness({ observe: async () => { throw new Error("unreadable"); } }, { strictTelemetry: true });
|
||||
let calls = 0;
|
||||
const outcome = controller.run("seat", "startup", async () => { calls++; }).catch(error => error);
|
||||
await flush(); expect(controller.status("seat")?.reason).toContain("strict telemetry");
|
||||
expect(controller.control("seat", "cancel")).toBe(true);
|
||||
expect((await outcome).code).toBe("startup_cancelled"); expect(calls).toBe(0);
|
||||
});
|
||||
|
||||
it("joins duplicate slow replies before waiting, preserving the original conversation arguments", async () => {
|
||||
const { controller } = harness(); const ready = deferred();
|
||||
const selected = { nativeId: "original", model: "chosen-model", effort: "chosen-effort" };
|
||||
const calls: unknown[] = [];
|
||||
const work = async () => { calls.push(structuredClone(selected)); await ready.promise; return selected; };
|
||||
const first = controller.run("seat", "restore", work);
|
||||
// The first caller has stopped waiting for its HTTP response; its native operation remains owned.
|
||||
const again = controller.run("seat", "restore", work);
|
||||
expect(again).toBe(first); await flush(); expect(calls).toEqual([selected]);
|
||||
ready.resolve(); expect(await again).toBe(selected); expect(await first).toBe(selected);
|
||||
});
|
||||
|
||||
it("revalidates a queued selection rather than starting a later occupant", async () => {
|
||||
let generation = "old"; const ready = deferred();
|
||||
const { controller } = harness({ selection: () => generation });
|
||||
const a = controller.run("a", "restore", async () => { await ready.promise; return { ok: true }; });
|
||||
let calls = 0;
|
||||
const b = controller.run("b", "restore", async () => { calls++; }).catch(error => error);
|
||||
await flush(); generation = "new"; ready.resolve(); await a;
|
||||
expect((await b).code).toBe("startup_selection_changed"); expect(calls).toBe(0);
|
||||
});
|
||||
|
||||
it("nested startup reuses the existing slot instead of deadlocking", async () => {
|
||||
const { controller } = harness();
|
||||
await expect(controller.run("seat", "restore", () => controller.run("seat", "startup", async () => "same native history")))
|
||||
.resolves.toBe("same native history");
|
||||
});
|
||||
|
||||
it.each(["live", "unknown"] as const)("reconciles a persisted %s launch before retry and never duplicates it", async state => {
|
||||
const previous: StartupAdmissionSnapshot = { attemptId: "interrupted", nodeId: "seat", kind: "restore", selection: "original-generation",
|
||||
phase: "starting", reason: "dispatched", observedAt: 1, dispatchedAt: 1, dispatchSeq: 5 };
|
||||
const { controller, records } = harness({ history: () => [previous], reconcile: async () => state });
|
||||
let calls = 0;
|
||||
await expect(controller.run("seat", "restore", async () => { calls++; })).rejects.toMatchObject({ code: "startup_unreconciled" });
|
||||
expect(calls).toBe(0); expect(controller.status("seat")?.attemptId).toBe("interrupted");
|
||||
expect(records).toEqual([]);
|
||||
});
|
||||
|
||||
it("recovery after proven process absence can retry, without treating absence as ready", async () => {
|
||||
const { controller } = harness({ history: () => [{ attemptId: "old", nodeId: "seat", kind: "restore", selection: "original-generation",
|
||||
phase: "uncertain", reason: "lost reply", observedAt: 1 }], reconcile: async () => "absent" });
|
||||
let calls = 0;
|
||||
await controller.run("seat", "restore", async () => { calls++; return { ok: true }; }); expect(calls).toBe(1);
|
||||
});
|
||||
|
||||
it("cancels during a pending OS observation without dispatching", async () => {
|
||||
const sample = deferred<StartupMemory>();
|
||||
const { controller } = harness({ observe: () => sample.promise });
|
||||
let calls = 0;
|
||||
const result = controller.run("seat", "restore", async () => { calls++; }).catch(error => error);
|
||||
expect(controller.control("seat", "cancel")).toBe(true);
|
||||
sample.resolve({ observedAt: 10000, source: "fixture", totalBytes: 1000, availableBytes: 800, pressure: "normal" });
|
||||
expect((await result).code).toBe("startup_cancelled"); await flush(); expect(calls).toBe(0);
|
||||
});
|
||||
|
||||
it("a refused preflight neither owns a native attempt nor blocks the next seat", async () => {
|
||||
const { controller, records } = harness();
|
||||
await expect(controller.run("a", "restore", async () => { throw new Error("Invalid projection"); })).rejects.toThrow("Invalid projection");
|
||||
await expect(controller.run("b", "startup", async () => "original session")).resolves.toBe("original session");
|
||||
expect(records.filter(r => r.nodeId === "a").some(r => r.phase === "uncertain")).toBe(false);
|
||||
});
|
||||
|
||||
it("cancellation during preparation is enforced at the native boundary", async () => {
|
||||
const { controller } = harness(); const prepare = deferred(); let nativeCalls = 0;
|
||||
const attempt = controller.run("seat", "restore", async () => {
|
||||
await prepare.promise;
|
||||
controller.beforeDispatch("seat", "exact-pane");
|
||||
nativeCalls++;
|
||||
}).catch(error => error);
|
||||
await flush(); expect(controller.status("seat")?.phase).toBe("preparing");
|
||||
expect(controller.control("seat", "cancel")).toBe(true);
|
||||
prepare.resolve(); expect((await attempt).code).toBe("startup_cancelled"); expect(nativeCalls).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("read-only startup memory interpretation", () => {
|
||||
it("uses Linux MemAvailable rather than MemFree", () => {
|
||||
expect(linuxStartupMemory("MemTotal: 1000 kB\nMemFree: 1 kB\nMemAvailable: 600 kB\n", 42))
|
||||
.toMatchObject({ totalBytes: 1024000, availableBytes: 614400, observedAt: 42 });
|
||||
});
|
||||
it("reads macOS query headroom and pressure independently", () => {
|
||||
expect(macStartupMemory("System-wide memory free percentage: 68%", "2\n", 1000, 42))
|
||||
.toMatchObject({ availableBytes: 680, pressure: "high" });
|
||||
expect(macStartupMemory("unrecognized output", "1", 1000, 42))
|
||||
.toMatchObject({ availableBytes: null, pressure: "normal" });
|
||||
});
|
||||
});
|
||||
@@ -155,6 +155,54 @@ describe("StartupOrchestrator", () => {
|
||||
};
|
||||
}
|
||||
|
||||
it("waits for shared startup admission before launching a native harness", async () => {
|
||||
let release!: () => void;
|
||||
const permit = new Promise<void>(resolve => { release = resolve; });
|
||||
const run = vi.fn(async (_nodeId: string, _kind: string, work: () => Promise<unknown>) => {
|
||||
await permit;
|
||||
return work();
|
||||
});
|
||||
Object.assign(tmux, { startupAdmission: { run, beforeDispatch: () => {} } });
|
||||
const adapter = mockAdapter();
|
||||
const attempt = createOrchestrator().startNode(makeInput(seedSession(), { adapter }));
|
||||
try {
|
||||
await Promise.resolve();
|
||||
expect(run).toHaveBeenCalledTimes(1);
|
||||
expect(adapter.launchHarness).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
release();
|
||||
await attempt;
|
||||
}
|
||||
expect(adapter.launchHarness).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("shares admission between independently constructed startup orchestrators", async () => {
|
||||
let release!: () => void;
|
||||
const permit = new Promise<void>(resolve => { release = resolve; });
|
||||
const run = vi.fn(async (_nodeId: string, _kind: string, work: () => Promise<unknown>) => {
|
||||
await permit;
|
||||
return work();
|
||||
});
|
||||
Object.assign(tmux, { startupAdmission: { run, beforeDispatch: () => {} } });
|
||||
const first = seedSession();
|
||||
const otherRig = rigRepo.createRig("other-rig");
|
||||
const otherNode = rigRepo.addNode(otherRig.id, "other", { runtime: "claude-code" });
|
||||
const otherSession = sessionRegistry.registerSession(otherNode.id, "other@other-rig");
|
||||
const second = { rigId: otherRig.id, nodeId: otherNode.id, sessionId: otherSession.id };
|
||||
const a = createOrchestrator().startNode(makeInput(first));
|
||||
const b = createOrchestrator().startNode(makeInput(second, {
|
||||
binding: { ...makeBinding(), nodeId: otherNode.id, tmuxSession: "other@other-rig" },
|
||||
}));
|
||||
try {
|
||||
await Promise.resolve();
|
||||
expect(run).toHaveBeenCalledTimes(2);
|
||||
expect(run.mock.calls.map(call => call[0])).toEqual([first.nodeId, second.nodeId]);
|
||||
} finally {
|
||||
release();
|
||||
await Promise.all([a, b]);
|
||||
}
|
||||
});
|
||||
|
||||
it("community contract: launch surfaces undelivered startup files without placing or pasting them", async () => {
|
||||
const root = nodeFs.mkdtempSync(nodePath.join(nodeOs.tmpdir(), "community-startup-files-"));
|
||||
try {
|
||||
|
||||
@@ -49,8 +49,8 @@ describe("SuccessorSessionLauncher", () => {
|
||||
return { runtime, launchHarness, checkReady } as unknown as RuntimeAdapter;
|
||||
}
|
||||
|
||||
function launcher(tmuxOptionDefaults?: TmuxOptionDefaultsApplier): SuccessorSessionLauncher {
|
||||
const tmux = { createSession, listPanes, killSession, respawnPane, setRemainOnExit, signalPaneProcess, isPaneDead, getDefaultShell, getPaneCommand } as unknown as TmuxAdapter;
|
||||
function launcher(tmuxOptionDefaults?: TmuxOptionDefaultsApplier, startupAdmission?: TmuxAdapter["startupAdmission"]): SuccessorSessionLauncher {
|
||||
const tmux = { startupAdmission, createSession, listPanes, killSession, respawnPane, setRemainOnExit, signalPaneProcess, isPaneDead, getDefaultShell, getPaneCommand } as unknown as TmuxAdapter;
|
||||
return new SuccessorSessionLauncher(tmux, discoveryRepo, {
|
||||
sessionEnv: { OPENRIG_HOME: "/home", HOME: "/daemon-home", CODEX_HOME: "/daemon-codex" },
|
||||
newId: () => "01ABCDEFG",
|
||||
@@ -63,6 +63,22 @@ describe("SuccessorSessionLauncher", () => {
|
||||
});
|
||||
}
|
||||
|
||||
it.each(["in-place", "staged"])("requires admission before a %s successor touches any native process", async mode => {
|
||||
const admission = { owns: () => false, run: vi.fn(async () => { throw new Error("Cancelled before dispatch"); }) } as unknown as NonNullable<TmuxAdapter["startupAdmission"]>;
|
||||
const instance = launcher(undefined, admission);
|
||||
const node = { id: "node-1", runtime: "codex", cwd: "/w" };
|
||||
const result = mode === "in-place"
|
||||
? await instance.createSuccessor({ node, departingSessionName: "seat@rig" })
|
||||
: await instance.stageSuccessor({ node, stagingSession: "staged@rig", canonicalSession: "seat@rig",
|
||||
occupantGeneration: "selected", onPaneCreated: vi.fn(), staged: {} as never });
|
||||
expect(result).toMatchObject({ ok: false, code: "startup_admission" });
|
||||
expect(admission.run).toHaveBeenCalledTimes(1);
|
||||
expect(createSession).not.toHaveBeenCalled();
|
||||
expect(signalPaneProcess).not.toHaveBeenCalled();
|
||||
expect(respawnPane).not.toHaveBeenCalled();
|
||||
expect(launchHarness).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("the successor's respawned pane gets the seat's declared env, and the declared names are recorded", async () => {
|
||||
const res = await launcher().createSuccessor({
|
||||
node: { id: "node-1", runtime: "codex", cwd: "/w", seatEnv: { MODE: "fast", REGION: "eu" } },
|
||||
|
||||
Reference in new issue
Block a user