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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions src/lib/onboard.ts
Original file line number Diff line number Diff line change
Expand Up @@ -484,9 +484,6 @@ const {
createInitialOnboardFlowPhases,
runInitialOnboardFlowSlice,
}: typeof import("./onboard/machine/initial-flow-phases") = require("./onboard/machine/initial-flow-phases");
const {
advanceTo,
}: typeof import("./onboard/machine/result") = require("./onboard/machine/result");
const { skippedStepMessage }: typeof import("./onboard/skipped-step-message") =
require("./onboard/skipped-step-message");
const policies: typeof import("./policy") = require("./policy");
Expand Down Expand Up @@ -4003,8 +4000,10 @@ const recordStateSkipped = onboardRuntimeBoundary.recordStateSkipped.bind(onboar
const recordRepairEvent = onboardRuntimeBoundary.recordRepairEvent.bind(onboardRuntimeBoundary);
const recordStateResult =
onboardRuntimeBoundary.recordStateResultWithStepCompatibility.bind(onboardRuntimeBoundary);
const recordCompatibleStateResult =
onboardRuntimeBoundary.recordCompatibleStateResult.bind(onboardRuntimeBoundary);
const recordInvalidatedStateResult =
onboardRuntimeBoundary.recordInvalidatedStateResult.bind(onboardRuntimeBoundary);
const recordInitialPreflightTransition =
onboardRuntimeBoundary.recordInitialPreflightTransition.bind(onboardRuntimeBoundary);
const recordPostVerifyStarted =
onboardRuntimeBoundary.recordPostVerifyStarted.bind(onboardRuntimeBoundary);

Expand Down Expand Up @@ -4230,9 +4229,7 @@ async function runOnboard(opts: OnboardOptions = {}): Promise<void> {
},
);
await onboardRuntimeBoundary.recordOnboardStarted(resume);
await (resume ? recordCompatibleStateResult : recordStateResult)(
advanceTo("preflight", { metadata: { state: "init" } }),
);
await recordInitialPreflightTransition(resume);
// Resume backstop: a session may exist without a sandboxName if sandbox
// creation failed before that step. Non-interactive --from cannot infer a
// safe name in that state.
Expand Down Expand Up @@ -4391,7 +4388,8 @@ async function runOnboard(opts: OnboardOptions = {}): Promise<void> {
runtime: onboardRuntimeBoundary.getRuntime(),
phases: [preflightPhase, gatewayPhase],
resume,
recordStateResult: recordCompatibleStateResult,
recordStateResult,
recordInvalidatedStateResult,
});

const initialContext = initialFlowResult.context;
Expand Down Expand Up @@ -4572,7 +4570,8 @@ async function runOnboard(opts: OnboardOptions = {}): Promise<void> {
runtime: onboardRuntimeBoundary.getRuntime(),
phases: [providerInferencePhase, sandboxPhase],
resume,
recordStateResult: recordCompatibleStateResult,
recordStateResult,
recordInvalidatedStateResult,
});
setupInferenceFactory.selectGatewayForFollowupOrExit(GATEWAY_NAME, runOpenshell);
const coreContext = coreFlowResult.context;
Expand Down Expand Up @@ -4727,7 +4726,8 @@ async function runOnboard(opts: OnboardOptions = {}): Promise<void> {
runtime: onboardRuntimeBoundary.getRuntime(),
phases: [branchSetupPhase, policiesPhase, finalizationPhase],
resume,
recordStateResult: recordCompatibleStateResult,
recordStateResult,
recordInvalidatedStateResult,
afterPoliciesResultApplied: () => {
sandboxCancelRollback.disarm();
},
Expand Down
25 changes: 14 additions & 11 deletions src/lib/onboard/machine/core-flow-phases.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,12 @@ function completeStep(): Session["steps"][string] {
};
}

function recordInvalidatedTargets(targets: string[]) {
return async (result: OnboardStateResult) => {
if (result.type === "transition") targets.push(result.next);
};
}

function createPhases(
overrides: {
providerDeps?: Partial<CoreOptions["providerDeps"]>;
Expand Down Expand Up @@ -478,6 +484,7 @@ describe("core onboard flow phases", () => {
recordStateResult: async (result) => {
if (result.type === "transition") recorded.push(result.next);
},
recordInvalidatedStateResult: recordInvalidatedTargets(recorded),
});

expect(recorded).toEqual(["sandbox", "openclaw"]);
Expand Down Expand Up @@ -519,6 +526,7 @@ describe("core onboard flow phases", () => {
recordStateResult: async (result) => {
recorded.push((result as ReturnType<typeof advanceTo>).next);
},
recordInvalidatedStateResult: recordInvalidatedTargets(recorded),
});

expect(recorded).toEqual(["sandbox", "openclaw"]);
Expand Down Expand Up @@ -610,17 +618,7 @@ describe("core onboard flow phases", () => {
resume: false,
recordStateResult: async (stateResult: OnboardStateResult) => {
if (stateResult.type !== "transition") return runtimeSession;
const source =
stateResult.metadata && typeof stateResult.metadata.state === "string"
? stateResult.metadata.state
: null;
if (
runtimeSession.machine.state === stateResult.next ||
source !== runtimeSession.machine.state
) {
skipped.push(`${source ?? "unknown"}->${stateResult.next}`);
return runtimeSession;
}
const source = stateResult.metadata?.state;
applied.push(`${source}->${stateResult.next}`);
runtimeSession = createSession({
machine: {
Expand All @@ -632,6 +630,11 @@ describe("core onboard flow phases", () => {
});
return runtimeSession;
},
recordInvalidatedStateResult: async (stateResult, invalidation) => {
if (stateResult.type !== "transition") return runtimeSession;
skipped.push(`${invalidation.sourceState ?? "unknown"}->${stateResult.next}`);
return runtimeSession;
},
});

expect(calls).toEqual(["provider_selection", "sandbox"]);
Expand Down
29 changes: 18 additions & 11 deletions src/lib/onboard/machine/core-flow-phases.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,10 @@ import {
type ProviderInferenceStateOptions,
} from "./handlers/provider-inference";
import { handleSandboxState, type SandboxStateOptions } from "./handlers/sandbox";
import { runLiveOnboardFlowSlice } from "./live-flow-slice";
import {
type InvalidatedOnboardStateResultRecorder,
runLiveOnboardFlowSlice,
} from "./live-flow-slice";
import type { OnboardStateResult } from "./result";
import type { OnboardMachineRunnerResult, OnboardMachineRunnerRuntime } from "./runner";
import type { OnboardSequencePhase } from "./sequence-runner";
Expand Down Expand Up @@ -163,20 +166,23 @@ export async function runCoreOnboardFlowSlice<Context extends OnboardFlowContext
phases: readonly OnboardSequencePhase<Context>[];
resume: boolean;
recordStateResult(result: OnboardStateResult): Promise<unknown>;
recordInvalidatedStateResult?: InvalidatedOnboardStateResultRecorder;
}): Promise<OnboardMachineRunnerResult<Context>> {
// Compatibility bridge for live resume repair when durable machine snapshots
// Recompute plan for live resume repair when durable machine snapshots
// are already downstream of this slice even though provider/sandbox
// repair/backstop checks must still re-run. Those ahead-state snapshots can
// come from legacy/test step mutation that explicitly opts into
// `updateMachine === true` or from repaired-resume replay of persisted
// sessions. This slice cannot eliminate that source locally because the
// repair/backstop checks are still modeled as imperative resume work rather
// than strict FSM recovery states. The tolerated downstream family includes
// sandbox branch states and the final slice handoff states: openclaw,
// agent_setup, policies, finalizing, and post_verify. Phase tests cover
// ahead-state resume and terminal-state rejection; remove this fallback once
// those checks are strict FSM recovery states and legacy machine step mutation
// is gone.
// sessions. Recomputed transition results are explicitly applied or
// invalidated by runLiveOnboardFlowSlice, so stale phase output cannot update
// context or silently advance state. This slice cannot eliminate that source
// locally because the repair/backstop checks are still modeled as imperative
// resume work rather than strict FSM recovery states. The tolerated downstream
// family includes sandbox branch states and the final slice handoff states:
// openclaw, agent_setup, policies, finalizing, and post_verify. Phase tests
// cover ahead-state resume and terminal-state rejection; remove this fallback
// once those checks are strict FSM recovery states and legacy machine step
// mutation is gone.
return runLiveOnboardFlowSlice({
context: options.context,
runtime: options.runtime,
Expand All @@ -195,6 +201,7 @@ export async function runCoreOnboardFlowSlice<Context extends OnboardFlowContext
]
: ["inference", "sandbox", "openclaw", "agent_setup"],
runSlice: runCoreOnboardFlowSequence,
applyCompatibleResult: options.recordStateResult,
recordStateResult: options.recordStateResult,
recordInvalidatedStateResult: options.recordInvalidatedStateResult,
});
}
50 changes: 33 additions & 17 deletions src/lib/onboard/machine/final-flow-phases.runtime.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,22 +24,27 @@ describe("final onboard flow runtime boundary", () => {
recordStepComplete: recorders.recordStepComplete,
recordPostVerifyStarted: recorders.recordPostVerifyStarted,
});
const compatibilityRecorder = vi.fn(
harness.boundary.recordCompatibleStateResult.bind(harness.boundary),
const recordStateResult = vi.fn(
harness.boundary.recordStateResultWithStepCompatibility.bind(harness.boundary),
);
const recordInvalidatedStateResult = vi.fn(
harness.boundary.recordInvalidatedStateResult.bind(harness.boundary),
);

await runFinalOnboardFlowSlice({
context: context({ session: harness.getSession() }),
runtime: harness.boundary.getRuntime(),
phases,
resume: false,
recordStateResult: compatibilityRecorder,
recordStateResult,
recordInvalidatedStateResult,
afterPoliciesResultApplied: () => {
order.push("disarm");
},
});

expect(compatibilityRecorder).not.toHaveBeenCalled();
expect(recordStateResult).not.toHaveBeenCalled();
expect(recordInvalidatedStateResult).not.toHaveBeenCalled();
expect(order).toEqual(["openclaw", "policies", "disarm", "set-default", "verify"]);
expect(harness.getSession()).toMatchObject({
status: "complete",
Expand All @@ -54,7 +59,7 @@ describe("final onboard flow runtime boundary", () => {
"policies",
"finalizing",
"post_verify",
] as const)("keeps persisted %s sessions on the compatibility path with the real runtime boundary", async (initialState) => {
] as const)("keeps persisted %s sessions on the recompute path with the real runtime boundary", async (initialState) => {
const order: string[] = [];
const harness = createRuntimeHarness(sessionAt(initialState));
const recorders = harness.boundary.recorders();
Expand All @@ -66,22 +71,27 @@ describe("final onboard flow runtime boundary", () => {
recordStepComplete: recorders.recordStepComplete,
recordPostVerifyStarted: recorders.recordPostVerifyStarted,
});
const compatibilityRecorder = vi.fn(
harness.boundary.recordCompatibleStateResult.bind(harness.boundary),
const recordStateResult = vi.fn(
harness.boundary.recordStateResultWithStepCompatibility.bind(harness.boundary),
);
const recordInvalidatedStateResult = vi.fn(
harness.boundary.recordInvalidatedStateResult.bind(harness.boundary),
);

await runFinalOnboardFlowSlice({
context: context({ session: harness.getSession() }),
runtime: harness.boundary.getRuntime(),
phases,
resume: false,
recordStateResult: compatibilityRecorder,
recordStateResult,
recordInvalidatedStateResult,
afterPoliciesResultApplied: () => {
order.push("disarm");
},
});

expect(compatibilityRecorder).toHaveBeenCalled();
expect(recordStateResult).toHaveBeenCalled();
expect(recordInvalidatedStateResult).toHaveBeenCalled();
expect(order).toEqual(["openclaw", "policies", "disarm", "set-default", "verify"]);
expect(harness.getSession()).toMatchObject({
status: "complete",
Expand All @@ -91,12 +101,12 @@ describe("final onboard flow runtime boundary", () => {
machine: { state: "complete" },
});

const skippedTargets = harness.events
.filter((event) => event.type === "state.result.skipped")
const invalidatedTargets = harness.events
.filter((event) => event.type === "state.result.invalidated")
.map((event) => event.metadata.targetState);
expect(skippedTargets).toContain("policies");
expect(invalidatedTargets).toContain("policies");
if (initialState !== "policies") {
expect(skippedTargets).toContain("finalizing");
expect(invalidatedTargets).toContain("finalizing");
}
});

Expand All @@ -112,22 +122,27 @@ describe("final onboard flow runtime boundary", () => {
recordStepComplete: recorders.recordStepComplete,
recordPostVerifyStarted: recorders.recordPostVerifyStarted,
});
const compatibilityRecorder = vi.fn(
harness.boundary.recordCompatibleStateResult.bind(harness.boundary),
const recordStateResult = vi.fn(
harness.boundary.recordStateResultWithStepCompatibility.bind(harness.boundary),
);
const recordInvalidatedStateResult = vi.fn(
harness.boundary.recordInvalidatedStateResult.bind(harness.boundary),
);

await runFinalOnboardFlowSlice({
context: context({ agent: { name: "hermes" }, session: harness.getSession() }),
runtime: harness.boundary.getRuntime(),
phases,
resume: false,
recordStateResult: compatibilityRecorder,
recordStateResult,
recordInvalidatedStateResult,
afterPoliciesResultApplied: () => {
order.push("disarm");
},
});

expect(compatibilityRecorder).not.toHaveBeenCalled();
expect(recordStateResult).not.toHaveBeenCalled();
expect(recordInvalidatedStateResult).not.toHaveBeenCalled();
expect(order).toEqual([
"agent-setup",
"agent-forward",
Expand Down Expand Up @@ -219,6 +234,7 @@ describe("final onboard flow runtime boundary", () => {
throw new Error("recording failed");
}
},
recordInvalidatedStateResult: vi.fn(),
afterPoliciesResultApplied: () => {
order.push("disarm");
},
Expand Down
3 changes: 3 additions & 0 deletions src/lib/onboard/machine/final-flow-phases.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ describe("final onboard flow phases", () => {
recorded.push(result.next);
}
},
recordInvalidatedStateResult: async (result) => {
if (result.type === "transition") recorded.push(result.next);
},
afterPoliciesResultApplied: () => {
order.push("disarm");
},
Expand Down
33 changes: 23 additions & 10 deletions src/lib/onboard/machine/final-flow-phases.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@ import { runFinalOnboardFlowSequence } from "./flow-slices";
import { type AgentSetupStateOptions, handleAgentSetupState } from "./handlers/agent-setup";
import { type FinalizationStateOptions, handleFinalizationState } from "./handlers/finalization";
import { handlePoliciesState, type PoliciesStateOptions } from "./handlers/policies";
import { runLiveOnboardFlowSlice } from "./live-flow-slice";
import {
type InvalidatedOnboardStateResultRecorder,
runLiveOnboardFlowSlice,
} from "./live-flow-slice";
import type { OnboardStateResult } from "./result";
import type { OnboardMachineRunnerRuntime } from "./runner";
import type { OnboardSequencePhase } from "./sequence-runner";
Expand Down Expand Up @@ -159,21 +162,24 @@ export async function runFinalOnboardFlowSlice<Context extends OnboardFlowContex
phases: readonly OnboardSequencePhase<Context>[];
resume: boolean;
recordStateResult(result: OnboardStateResult): Promise<unknown>;
recordInvalidatedStateResult?: InvalidatedOnboardStateResultRecorder;
afterPoliciesResultApplied?(): void;
onContextUpdated?(context: Context): void;
}): Promise<void> {
// Compatibility bridge for live resume repair when durable machine snapshots
// Recompute plan for live resume repair when durable machine snapshots
// are already downstream of this slice even though branch setup/readiness,
// policy reconciliation, and final verification must still re-run. Those
// ahead-state snapshots can come from legacy/test step mutation that
// explicitly opts into `updateMachine === true` or from repaired-resume replay
// of persisted sessions. This slice cannot eliminate that source locally
// because final-phase repair checks are still modeled as imperative resume
// work rather than strict FSM recovery states. The tolerated downstream states
// are "policies", "finalizing", and "post_verify". Phase tests cover
// ahead-state resume and terminal-state rejection; remove this fallback once
// final-phase repair checks are first-class FSM recovery states and legacy
// machine step mutation is gone.
// of persisted sessions. Recomputed transition results are explicitly applied
// or invalidated by runLiveOnboardFlowSlice, so stale phase output cannot
// update context or silently advance state. This slice cannot eliminate that
// source locally because final-phase repair checks are still modeled as
// imperative resume work rather than strict FSM recovery states. The tolerated
// downstream states are "policies", "finalizing", and "post_verify". Phase
// tests cover ahead-state resume and terminal-state rejection; remove this
// fallback once final-phase repair checks are first-class FSM recovery states
// and legacy machine step mutation is gone.
await runLiveOnboardFlowSlice({
context: options.context,
runtime: withAfterPoliciesResultApplied(options.runtime, options.afterPoliciesResultApplied),
Expand All @@ -183,9 +189,16 @@ export async function runFinalOnboardFlowSlice<Context extends OnboardFlowContex
? ["openclaw", "agent_setup", "policies", "finalizing", "post_verify"]
: ["policies", "finalizing", "post_verify"],
runSlice: runFinalOnboardFlowSequence,
applyCompatibleResult: async (stateResult) => {
recordStateResult: async (stateResult) => {
await options.recordStateResult(stateResult);
if (isPoliciesAppliedResult(stateResult)) options.afterPoliciesResultApplied?.();
},
recordInvalidatedStateResult: async (stateResult, invalidation) => {
if (!options.recordInvalidatedStateResult) {
throw new Error("Missing onboarding state result invalidation recorder");
}
await options.recordInvalidatedStateResult(stateResult, invalidation);
if (isPoliciesAppliedResult(stateResult)) options.afterPoliciesResultApplied?.();
},
});
}
Loading
Loading