History

fix(orchestration): preserve v2 thread deletion planning

Julius Marminge committed 2 months ago

Commit 04bbc331 parent 4b245473 files · +28 −26Browse files

Unified · Split

3 files
  1. apps/server/src/orchestration-v2/Orchestrator.ts +13 −21
  2. apps/server/src/orchestration-v2/runtimeLayer.test.ts +14 −5
  3. apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +1 −0
apps/server/src/orchestration-v2/Orchestrator.ts+13−21
@@ -1912,9 +1912,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 1911 unmodified lines
1912 // decided serially against the projection, so a turn start that1912 // decided serially against the projection, so a turn start that
1913 // re-engages the thread cannot race this detach.1913 // re-engages the thread cannot race this detach.
1914 const detachSessionIds = new Set(1914 const detachSessionIds = new Set(
1915 command.type === "thread.archive" ||1915 command.type === "thread.archive" || command.type === "thread.settle"
1916 command.type === "thread.delete" ||
1917 command.type === "thread.settle"
1918 ? projection.providerSessions.map((session) => session.id)1916 ? projection.providerSessions.map((session) => session.id)
1919 : command.type === "thread.metadata.update" &&1917 : command.type === "thread.metadata.update" &&
1920 command.worktreePath !== undefined &&1918 command.worktreePath !== undefined &&
@@ -1956,13 +1954,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 35 unmodified lines
1956 ? "Thread archived."1954 ? "Thread archived."
1957 : command.type === "thread.settle"1955 : command.type === "thread.settle"
1958 ? "Thread settled."1956 ? "Thread settled."
1959 : command.type === "thread.delete"1957 : command.type === "thread.metadata.update"
1960 ? "Thread deleted."1958 ? "Workspace changed."
1961 : command.type === "thread.metadata.update"1959 : command.type === "thread.runtime-mode.set"
1962 ? "Workspace changed."1960 ? "Runtime mode changed."
1963 : command.type === "thread.runtime-mode.set"1961 : "Provider or model selection changed.",
1964 ? "Runtime mode changed."
1965 : "Provider or model selection changed.",
1966 },1962 },
1967 });1963 });
1968 const pendingEffect = {1964 const pendingEffect = {
@@ -1977,19 +1973,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 8 unmodified lines
1977 ? "Thread archived."1973 ? "Thread archived."
1978 : command.type === "thread.settle"1974 : command.type === "thread.settle"
1979 ? "Thread settled."1975 ? "Thread settled."
1980 : command.type === "thread.delete"1976 : command.type === "thread.metadata.update"
1981 ? "Thread deleted."1977 ? "Workspace changed."
1982 : command.type === "thread.metadata.update"1978 : command.type === "thread.runtime-mode.set"
1983 ? "Workspace changed."1979 ? "Runtime mode changed."
1984 : command.type === "thread.runtime-mode.set"1980 : "Provider or model selection changed.",
1985 ? "Runtime mode changed."
1986 : "Provider or model selection changed.",
1987 // Terminal detaches revoke the thread's MCP credentials; other1981 // Terminal detaches revoke the thread's MCP credentials; other
1988 // detach reasons keep them so a re-attaching provider process1982 // detach reasons keep them so a re-attaching provider process
1989 // stays authorized.1983 // stays authorized.
1990 ...(command.type === "thread.archive" || command.type === "thread.delete"1984 ...(command.type === "thread.archive" ? { revokeMcpCredential: true } : {}),
1991 ? { revokeMcpCredential: true }
1992 : {}),
1993 },1985 },
1994 } satisfies PendingOrchestrationEffectV2;1986 } satisfies PendingOrchestrationEffectV2;
1995 yield* Ref.update(effects, (existing) => [...existing, pendingEffect]);1987 yield* Ref.update(effects, (existing) => [...existing, pendingEffect]);
@@ -1998,7 +1990,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 2 unmodified lines
1998 );1990 );
1999 }1991 }
2000 1992
2001 if (command.type === "thread.archive" || command.type === "thread.delete") {1993 if (command.type === "thread.archive") {
2002 yield* Ref.update(effects, (existing) => [1994 yield* Ref.update(effects, (existing) => [
2003 ...existing,1995 ...existing,
2004 {1996 {
apps/server/src/orchestration-v2/runtimeLayer.test.ts+14−5
@@ -24,6 +24,7 @@ import * as Effect from "effect/Effect"; · 23 unmodified lines
24import * as DateTime from "effect/DateTime";24import * as DateTime from "effect/DateTime";
25import * as Deferred from "effect/Deferred";25import * as Deferred from "effect/Deferred";
26import * as Fiber from "effect/Fiber";26import * as Fiber from "effect/Fiber";
27import * as FileSystem from "effect/FileSystem";
27import * as Layer from "effect/Layer";28import * as Layer from "effect/Layer";
28import * as Option from "effect/Option";29import * as Option from "effect/Option";
29import * as Queue from "effect/Queue";30import * as Queue from "effect/Queue";
@@ -908,7 +909,7 @@ it.layer(LegacyImportTestLayer)("OrchestrationV2 legacy import", (it) => { · 878 unmodified lines
908 );909 );
909});910});
910 911
911it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {912it.layer(Layer.merge(TestLayer, NodeServices.layer))("OrchestrationV2LayerLive lifecycle", (it) => {
912 it.effect("applies lifecycle commands idempotently and emits archive/removal shell deltas", () =>913 it.effect("applies lifecycle commands idempotently and emits archive/removal shell deltas", () =>
913 Effect.gen(function* () {914 Effect.gen(function* () {
914 const orchestrator = yield* OrchestratorV2;915 const orchestrator = yield* OrchestratorV2;
@@ -1474,6 +1475,10 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 559 unmodified lines
1474 1475
1475 it.effect("applies deferred settlement after safe completion and replays its receipt", () =>1476 it.effect("applies deferred settlement after safe completion and replays its receipt", () =>
1476 Effect.gen(function* () {1477 Effect.gen(function* () {
1478 const fileSystem = yield* FileSystem.FileSystem;
1479 const workspacePath = yield* fileSystem.makeTempDirectoryScoped({
1480 prefix: "t3-runtime-layer-deferred-settle-",
1481 });
1477 const orchestrator = yield* OrchestratorV2;1482 const orchestrator = yield* OrchestratorV2;
1478 const eventSink = yield* EventSinkV2;1483 const eventSink = yield* EventSinkV2;
1479 const sql = yield* SqlClient.SqlClient;1484 const sql = yield* SqlClient.SqlClient;
@@ -1494,7 +1499,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 14 unmodified lines
1494 runtimeMode: "full-access",1499 runtimeMode: "full-access",
1495 interactionMode: "default",1500 interactionMode: "default",
1496 branch: null,1501 branch: null,
1497 worktreePath: "/tmp/runtime-layer-deferred-settle",1502 worktreePath: workspacePath,
1498 });1503 });
1499 yield* orchestrator.dispatch({1504 yield* orchestrator.dispatch({
1500 type: "thread.pin",1505 type: "thread.pin",
@@ -1525,7 +1530,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 24 unmodified lines
1525 providerSessionId,1530 providerSessionId,
1526 modelSelection,1531 modelSelection,
1527 runtimePolicy: {1532 runtimePolicy: {
1528 cwd: "/tmp/runtime-layer-deferred-settle",1533 cwd: workspacePath,
1529 runtimeMode: "full-access",1534 runtimeMode: "full-access",
1530 interactionMode: "default",1535 interactionMode: "default",
1531 },1536 },
@@ -1631,6 +1636,10 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 99 unmodified lines
1631 1636
1632 it.effect("archives after safe completion and discards stale or blocked deferred intents", () =>1637 it.effect("archives after safe completion and discards stale or blocked deferred intents", () =>
1633 Effect.gen(function* () {1638 Effect.gen(function* () {
1639 const fileSystem = yield* FileSystem.FileSystem;
1640 const workspacePath = yield* fileSystem.makeTempDirectoryScoped({
1641 prefix: "t3-runtime-layer-deferred-organization-",
1642 });
1634 const orchestrator = yield* OrchestratorV2;1643 const orchestrator = yield* OrchestratorV2;
1635 const eventSink = yield* EventSinkV2;1644 const eventSink = yield* EventSinkV2;
1636 const sql = yield* SqlClient.SqlClient;1645 const sql = yield* SqlClient.SqlClient;
@@ -1652,7 +1661,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 15 unmodified lines
1652 runtimeMode: "full-access",1661 runtimeMode: "full-access",
1653 interactionMode: "default",1662 interactionMode: "default",
1654 branch: null,1663 branch: null,
1655 worktreePath: `/tmp/runtime-layer-deferred-${suffix}`,1664 worktreePath: workspacePath,
1656 });1665 });
1657 yield* orchestrator.dispatch({1666 yield* orchestrator.dispatch({
1658 type: "message.dispatch",1667 type: "message.dispatch",
@@ -1716,7 +1725,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 57 unmodified lines
1716 providerSessionId,1725 providerSessionId,
1717 modelSelection,1726 modelSelection,
1718 runtimePolicy: {1727 runtimePolicy: {
1719 cwd: "/tmp/runtime-layer-deferred-archive",1728 cwd: workspacePath,
1720 runtimeMode: "full-access",1729 runtimeMode: "full-access",
1721 interactionMode: "default",1730 interactionMode: "default",
1722 },1731 },
apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts+1−0
@@ -427,6 +427,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry<Error>( · 426 unmodified lines
427 Layer.provide(427 Layer.provide(
428 Layer.mergeAll(effectWorkerProvided, storesLayer, eventSinkProvided, idAllocatorLayer),428 Layer.mergeAll(effectWorkerProvided, storesLayer, eventSinkProvided, idAllocatorLayer),
429 ),429 ),
430 Layer.provide(serverSettingsLayer),
430 );431 );
431 const replayRuntime = Layer.mergeAll(432 const replayRuntime = Layer.mergeAll(
432 orchestratorProvided,433 orchestratorProvided,