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
418 changes: 417 additions & 1 deletion packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts

Large diffs are not rendered by default.

4 changes: 2 additions & 2 deletions packages/runtime-host/src/server/execution-composition.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1240,8 +1240,8 @@ export async function createExecutionRuntimeHostComposition(
bindRun: (identity) => messages.bindRun(identity),
executeRoot: (input) =>
executeHostedExecutionToSettlement(requireRootCoordinator(rootCoordinator), input),
stopRoot: (identity, input) =>
requireRootCoordinator(rootCoordinator).stopRoot(identity, input),
stopRoot: (identity, input, options) =>
requireRootCoordinator(rootCoordinator).stopRoot(identity, input, options),
stopSession: (sessionId, input) =>
requireRootCoordinator(rootCoordinator).stopSession(sessionId, input),
};
Expand Down
49 changes: 44 additions & 5 deletions packages/runtime-host/src/server/root-turn-coordinator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ import {
RuntimeHostedRootConflictError,
RuntimeHostedRootUnavailableError,
RuntimeMessageAuthorityInvariantError,
type RuntimeHostedRootStopOptions,
type RuntimeMessageRunIdentity,
} from '@maka/runtime/message-authority';
import {
Expand Down Expand Up @@ -1240,15 +1241,24 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
);
}

stopRoot(identity: RuntimeMessageRunIdentity, input: StopSessionInput = {}): Promise<void> {
stopRoot(
identity: RuntimeMessageRunIdentity,
input: StopSessionInput = {},
options: RuntimeHostedRootStopOptions = {},
): Promise<void> {
normalizeStopSessionSource(input.source, input.workHubActionId);
// Only an activation-scoped graph stop targets one Run. Public turn.stop,
// shutdown and supervisor stops keep the Session-level Runtime stop, which
// also reaches a claim whose Run has not attached yet.
const scoped = options.scope === 'run';
return this.runCommand(async () => {
const declared = await this.sessionAdmission.run(identity.sessionId, (lease) =>
this.declareStopFence(
identity,
() => this.messages.commitStopFence(identity),
lease,
input,
scoped,
),
);
await declared?.deliverStop();
Expand All @@ -1258,10 +1268,28 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
);
if (disposition.kind === 'complete') {
if (!disposition.outcome.ok) throwHostedStopError(identity.sessionId, disposition.outcome);
if (!scoped) return;
const logical = await readLogicalRuntimeExecutionForRun(
this.stores.runtimeEventStore,
identity,
);
await this.deliverRuntimeStopIntent(identity.sessionId, input, {
...identity,
runId: logical?.tip.runId ?? identity.runId,
});
return;
}
if (disposition.kind === 'request_stop') {
await this.deliverRuntimeStopIntent(identity.sessionId, input);
await this.deliverRuntimeStopIntent(
identity.sessionId,
input,
scoped
? {
...identity,
runId: disposition.active.continuation?.runId ?? identity.runId,
}
: undefined,
);
}
await disposition.active.done;
});
Expand Down Expand Up @@ -2700,6 +2728,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
commitQueueFence: () => QueueFenceResult,
admission: SessionAdmissionLease,
stopInput: StopSessionInput = {},
scoped = false,
): Promise<DeclaredStopFence | undefined> {
const active = this.#executions.get(input.sessionId);
if (!active || active.turnId !== input.turnId || active.runId !== input.runId) {
Expand All @@ -2718,13 +2747,22 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
'turn_stopped',
admission,
);
const shouldDeliverStop = !active.stopRequested;
const shouldDeliverStop = scoped || !active.stopRequested;
active.stopRequested = stopInput;
return {
active,
deliverStop: () =>
shouldDeliverStop
? this.deliverRuntimeStopIntent(input.sessionId, stopInput)
? this.deliverRuntimeStopIntent(
input.sessionId,
stopInput,
scoped
? {
...input,
runId: active.continuation?.runId ?? active.runId,
}
: undefined,
)
: Promise.resolve(),
};
}
Expand Down Expand Up @@ -3376,8 +3414,9 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
private async deliverRuntimeStopIntent(
sessionId: string,
input: StopSessionInput = { source: 'stop_button' },
identity?: RuntimeMessageRunIdentity,
): Promise<void> {
await this.manager.deliverHostedRootStop(sessionId, input);
await this.manager.deliverHostedRootStop(sessionId, input, identity);
}

private async stopActiveTurn(sessionId: string, active: ActiveRootTurn): Promise<void> {
Expand Down
Loading