Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
152 changes: 152 additions & 0 deletions apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import * as Layer from "effect/Layer";
import * as SqlClient from "effect/unstable/sql/SqlClient";
import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
import * as EffectOutbox from "./EffectOutbox.ts";
import * as Orchestrator from "./Orchestrator.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
Expand All @@ -42,13 +43,164 @@ const database = SqlitePersistenceMemory;
const testLayer = Layer.mergeAll(
database,
ProjectionStore.layer.pipe(Layer.provide(database)),
EffectOutbox.layer.pipe(Layer.provide(database)),
makeOrchestratorV2ReplayLayerWithRegistry(
{ name: "control-reads" },
ProviderAdapterRegistry.makeLayer([adapter]),
{ databaseLayer: database, runEffectWorker: false },
),
);

it.effect("interrupts only the selected running native Codex subagent", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const projections = yield* ProjectionStore.ProjectionStoreV2;
const outbox = yield* EffectOutbox.EffectOutboxV2;
const parentThreadId = ThreadId.make("parent:stop-subagent");
const childThreadId = ThreadId.make("child:stop-subagent");
const subagentId = NodeId.make("subagent:stop-subagent");
const providerThreadId = ProviderThreadId.make("provider-thread:stop-subagent");
const providerTurnId = ProviderTurnId.make("provider-turn:stop-subagent");
const providerSessionId = ProviderSessionId.make("session:stop-subagent");
const now = yield* DateTime.now;
for (const threadId of [parentThreadId, childThreadId]) {
yield* orchestrator.dispatch({
type: "thread.create",
commandId: CommandId.make(`create:${threadId}`),
threadId,
projectId: ProjectId.make("project:stop-subagent"),
title: "Agent",
modelSelection,
runtimeMode: "full-access",
interactionMode: "default",
branch: null,
worktreePath: null,
createdBy: "user",
creationSource: "web",
});
}
const subagent = {
id: subagentId,
threadId: parentThreadId,
runId: null,
parentNodeId: NodeId.make("root:stop-subagent"),
origin: "provider_native" as const,
createdBy: "agent" as const,
driver: adapter.driver,
providerInstanceId: instanceId,
providerThreadId: null,
childThreadId,
nativeTaskRef: null,
prompt: "Check this",
title: "Worker",
model: "gpt-5.1-codex",
status: "running" as const,
result: null,
startedAt: now,
completedAt: null,
updatedAt: now,
};
yield* projections.apply({
id: EventId.make("subagent:stop-subagent"),
type: "subagent.updated",
threadId: parentThreadId,
occurredAt: now,
payload: subagent,
});
yield* projections.apply({
id: EventId.make("provider-thread:stop-subagent"),
type: "provider-thread.updated",
threadId: childThreadId,
occurredAt: now,
payload: {
id: providerThreadId,
driver: adapter.driver,
providerInstanceId: instanceId,
providerSessionId,
appThreadId: childThreadId,
ownerNodeId: subagentId,
nativeThreadRef: null,
nativeConversationHeadRef: null,
status: "active",
firstRunOrdinal: null,
lastRunOrdinal: null,
handoffIds: [],
forkedFrom: null,
createdAt: now,
updatedAt: now,
},
});
yield* projections.apply({
id: EventId.make("provider-turn:stop-subagent"),
type: "provider-turn.updated",
threadId: childThreadId,
occurredAt: now,
payload: {
id: providerTurnId,
providerThreadId,
nodeId: NodeId.make("root:child-stop-subagent"),
runAttemptId: null,
nativeTurnRef: null,
ordinal: 1,
status: "running",
startedAt: now,
completedAt: null,
},
});
const commandId = CommandId.make("interrupt:stop-subagent");
const accepted = yield* orchestrator.dispatch({
type: "subagent.interrupt",
commandId,
threadId: parentThreadId,
subagentId,
});
assert.deepEqual(
accepted.storedEvents.map((event) => event.event.type),
["subagent.interrupt-requested"],
);
assert.equal(
(yield* projections.getThreadProjection(parentThreadId)).subagents[0]?.status,
"running",
);
assert.deepEqual(
(yield* outbox.listByCommandId(commandId)).map(({ threadId, request }) => ({
threadId,
request,
})),
[
{
threadId: childThreadId,
request: {
type: "provider-turn.interrupt",
providerSessionId,
providerThreadId,
providerTurnId,
},
},
],
);
yield* projections.apply({
id: EventId.make("subagent:stop-subagent:completed"),
type: "subagent.updated",
threadId: parentThreadId,
occurredAt: now,
payload: { ...subagent, status: "completed", completedAt: now },
});
const settledCommandId = CommandId.make("interrupt:stop-subagent:settled");
const rejected = yield* orchestrator
.dispatch({
type: "subagent.interrupt",
commandId: settledCommandId,
threadId: parentThreadId,
subagentId,
})
.pipe(Effect.flip);
assert.instanceOf(rejected, Orchestrator.OrchestratorDispatchError);
assert.equal(rejected.cause, undefined);
assert.deepEqual(yield* outbox.listByCommandId(settledCommandId), []);
}).pipe(Effect.provide(testLayer)),
);

it.effect(
"dispatches metadata, queue resume and request controls without hydrating unrelated history",
() =>
Expand Down
69 changes: 69 additions & 0 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId {
case "prepared-run.progress":
case "prepared-run.fail":
case "run.interrupt":
case "subagent.interrupt":
case "queued-message.promote-to-steer":
case "queue.resume":
case "queued-run.reorder":
Expand Down Expand Up @@ -8204,6 +8205,71 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
return undefined;
});

const dispatchSubagentInterrupt = (
command: Extract<OrchestrationV2Command, { readonly type: "subagent.interrupt" }>,
events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
) =>
Effect.gen(function* () {
const parent = yield* loadProjectionForCommand(command, ["subagents"]);
const subagent = parent.subagents.find((entry) => entry.id === command.subagentId);
if (
subagent?.origin !== "provider_native" ||
subagent.driver !== "codex" ||
subagent.status !== "running" ||
subagent.childThreadId === null
) {
return yield* new OrchestratorDispatchError({
commandId: command.commandId,
commandType: command.type,
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
}
const childThreadId = subagent.childThreadId;
const child = yield* projectionStore
.getThreadRecords(childThreadId, ["providerThreads", "providerTurns"])
.pipe(
Effect.mapError(
(cause) => new OrchestratorProjectionError({ threadId: childThreadId, cause }),
),
);
const turn = child.providerTurns.findLast((entry) => entry.status === "running");
const providerThread = child.providerThreads.find(
(entry) => entry.id === turn?.providerThreadId,
);
if (turn === undefined || providerThread?.providerSessionId == null) {
Comment thread
Bil0000 marked this conversation as resolved.
Outdated
return yield* new OrchestratorDispatchError({
commandId: command.commandId,
commandType: command.type,
});
}
const providerSessionId = providerThread.providerSessionId;
const emitEvent = emit(events, command);
yield* emitEvent({
type: "subagent.interrupt-requested",
threadId: command.threadId,
...(subagent.runId === null ? {} : { runId: subagent.runId }),
nodeId: subagent.id,
driver: subagent.driver,
providerInstanceId: subagent.providerInstanceId,
occurredAt: yield* DateTime.now,
payload: subagent.id,
});
yield* Ref.update(effects, (existing) => [
...existing,
{
id: `effect:${command.commandId}:provider-turn.interrupt:${turn.id}`,
commandId: command.commandId,
threadId: childThreadId,
request: {
type: "provider-turn.interrupt",
providerSessionId,
providerThreadId: providerThread.id,
providerTurnId: turn.id,
},
} satisfies PendingOrchestrationEffectV2,
]);
});

const dispatchCheckpointRollback = (
command: Extract<OrchestrationV2Command, { readonly type: "checkpoint.rollback" }>,
events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
Expand Down Expand Up @@ -9332,6 +9398,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
case "run.interrupt":
cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects);
break;
case "subagent.interrupt":
yield* dispatchSubagentInterrupt(command, events, effects);
break;
case "queued-message.promote-to-steer":
yield* dispatchQueuedMessagePromoteToSteer(command, events, effects);
break;
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -638,6 +638,8 @@ export function applyToProjection(
};

switch (event.type) {
case "subagent.interrupt-requested":
return projection;
case "thread.created":
case "thread.archived":
case "thread.unarchived":
Expand Down Expand Up @@ -1663,6 +1665,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
const apply: ProjectionStoreV2Shape["apply"] = (event) =>
Effect.gen(function* () {
switch (event.type) {
case "subagent.interrupt-requested":
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
break;
case "thread.created":
case "thread.archived":
case "thread.unarchived":
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray<Thread
case "prepared-run.progress":
case "prepared-run.fail":
case "run.interrupt":
case "subagent.interrupt":
case "queued-message.promote-to-steer":
case "queue.resume":
case "queued-run.reorder":
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/relay/AgentAwarenessRelay.ts
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,7 @@ export function shouldPublishAgentAwarenessEvent(
case "run-attempt.created":
case "run-attempt.updated":
case "node.updated":
case "subagent.interrupt-requested":
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
case "provider-session.attached":
case "provider-session.updated":
case "provider-session.detached":
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ const state = vi.hoisted(() => ({
projects: [] as unknown[],
configs: new Map<string, unknown>(),
showTooltips: false,
command: vi.fn().mockResolvedValue({ _tag: "Success" }),
}));

vi.mock("@tanstack/react-router", () => ({ useNavigate: () => state.navigate }));
Expand All @@ -23,7 +24,7 @@ vi.mock("../../state/entities", () => ({
vi.mock("../../lib/archivedThreadsState", () => ({
useArchivedThreadSnapshots: () => ({ snapshots: [] }),
}));
vi.mock("../../state/use-atom-command", () => ({ useAtomCommand: () => vi.fn() }));
vi.mock("../../state/use-atom-command", () => ({ useAtomCommand: () => state.command }));
vi.mock("../ui/tooltip", () => ({
Tooltip: ({ children }: { children: ReactNode }) => children,
TooltipTrigger: ({ render, children }: { render: ReactElement; children: ReactNode }) =>
Expand All @@ -42,6 +43,83 @@ afterEach(async () => {
state.projects = [];
state.configs.clear();
state.showTooltips = false;
state.command.mockClear();
state.projection = null;
});

it("stops only active subagents from lineage without opening their thread", async () => {
vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true);
const parent = {
id: "parent",
lineage: { relationshipToParent: null },
activeProviderThreadId: null,
};
const child = {
id: "child",
title: "Worker",
lineage: { parentThreadId: "parent", relationshipToParent: "subagent" },
};
const agent = {
id: "agent",
childThreadId: "child",
origin: "app_owned",
driver: "codex",
providerInstanceId: "codex",
title: "Worker",
prompt: "Check the change",
model: "gpt-5.4",
status: "running",
progress: null,
result: null,
startedAt: DateTime.makeUnsafe("2026-09-16T12:00:00Z"),
completedAt: null,
updatedAt: DateTime.makeUnsafe("2026-09-16T12:00:00Z"),
};
state.shells = [{ environmentId: "test", source: child }];
const projection = {
thread: parent,
runs: [],
providerThreads: [],
providerSessions: [],
contextTransfers: [],
subagents: [agent],
};
state.projection = projection;
const panel = (
<ThreadRelationshipsPanel
environmentId={EnvironmentId.make("test")}
threadId={ThreadId.make("parent")}
/>
);
await act(async () => {
renderer = create(panel);
});
const stopButton = () => renderer.root.findByProps({ "aria-label": "Stop subagent Worker" });
await act(async () => stopButton().props.onClick());
expect(state.command).toHaveBeenCalledWith({
environmentId: "test",
input: { threadId: "child" },
});
expect(state.navigate).not.toHaveBeenCalled();

state.command.mockClear();
state.projection = { ...projection, subagents: [{ ...agent, origin: "provider_native" }] };
await act(async () => renderer.update(cloneElement(panel)));
await act(async () => stopButton().props.onClick());
expect(state.command).toHaveBeenCalledWith({
environmentId: "test",
input: { threadId: "parent", subagentId: "agent" },
});

state.projection = { ...projection, subagents: [{ ...agent, status: "completed" }] };
await act(async () => renderer.update(cloneElement(panel)));
expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(0);
state.projection = {
...projection,
subagents: [{ ...agent, origin: "provider_native", driver: "claudeAgent" }],
};
await act(async () => renderer.update(cloneElement(panel)));
expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(0);
});

it("shows the matching child agent details and refreshes them when the agent settles", async () => {
Expand Down
Loading
Loading