lael-labs/t3code

.plans/21-orchestration-v2-application-integration.md+2−0

@@ -292,6 +292,8 @@ Add coverage for: · 291 unmodified lines
292		292
293	## Shape 2: Application Services Around V2	293	## Shape 2: Application Services Around V2
294		294
		295	Implementation status: complete. The production WebSocket/client cutover remains intentionally deferred to Shape 3.
		296
295	### Goal	297	### Goal
296		298
297	Recreate the application behavior currently hidden in V1 reactors and WebSocket handlers using explicit V2 services and durable workflows.	299	Recreate the application behavior currently hidden in V1 reactors and WebSocket handlers using explicit V2 services and durable workflows.

apps/server/src/orchestration-v2/EffectOutbox.ts+17−6

     checkpointId: CheckpointId,
     scopeId: CheckpointScopeId,
   }),
-  Schema.Struct({
-    type: Schema.Literal("provider-thread.fork"),
-    sourceProviderThreadId: ProviderThreadId,
-    targetProviderThreadId: ProviderThreadId,
-  }),
   Schema.Struct({
     type: Schema.Literal("checkpoint.capture"),
     runId: RunId,
   }),
   Schema.Struct({
     type: Schema.Literal("terminal.cleanup"),
-    runId: RunId,
   }),
   Schema.Struct({
     type: Schema.Literal("attachment.cleanup"),
   readonly listByCommandId: (
     commandId: CommandId,
   ) => Effect.Effect<ReadonlyArray<OrchestrationEffectV2>, EffectOutboxError>;
+  readonly reclaimRunning: Effect.Effect<number, EffectOutboxError>;
   readonly claimNext: (input: {
     readonly workerId: string;
     readonly leaseDurationMs: number;
               : new EffectOutboxError({ operation: "list", cause }),
           ),
         ),
+      reclaimRunning: Effect.gen(function* () {
+        const now = DateTime.formatIso(yield* DateTime.now);
+        const rows = yield* sql<{ readonly effect_id: string }>`
+          UPDATE orchestration_v2_effect_outbox
+          SET
+            status = 'pending',
+            lease_owner = NULL,
+            lease_expires_at = NULL,
+            available_at = ${now},
+            updated_at = ${now}
+          WHERE status = 'running'
+          RETURNING effect_id
+        `;
+        if (rows.length > 0) yield* Queue.offer(available, undefined);
+        return rows.length;
+      }).pipe(Effect.mapError((cause) => new EffectOutboxError({ operation: "reclaim", cause }))),
       claimNext: ({ workerId, leaseDurationMs }) =>
         Effect.gen(function* () {
           const now = yield* DateTime.now;

apps/server/src/orchestration-v2/EffectWorker.ts+26−12

 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
-import { CheckpointCaptureServiceV2 } from "./CheckpointCaptureService.ts";
+import { RunFinalizationService } from "./RunFinalizationService.ts";
+import { ResourceCleanupService } from "./ResourceCleanupService.ts";
 import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts";
 import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
 import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
   OrchestrationEffectExecutorV2,
   never,
   | ProviderSessionManagerV2
-  | CheckpointCaptureServiceV2
+  | RunFinalizationService
   | CheckpointRollbackServiceV2
   | ProviderTurnControlServiceV2
   | ProviderTurnStartServiceV2
 > = Layer.effect(
   OrchestrationEffectExecutorV2,
   Effect.gen(function* () {
-    const checkpointCapture = yield* CheckpointCaptureServiceV2;
+    const runFinalization = yield* RunFinalizationService;
+    const resourceCleanup = yield* ResourceCleanupService;
     const checkpointRollback = yield* CheckpointRollbackServiceV2;
     const providerSessions = yield* ProviderSessionManagerV2;
     const providerTurnControl = yield* ProviderTurnControlServiceV2;
                 ),
               );
           case "checkpoint.capture":
-            return checkpointCapture
+            return runFinalization
-              .execute({
+              .finalize({
                 threadId: effect.threadId,
                 runId: effect.request.runId,
                 scopeId: effect.request.scopeId,
                     }),
                 ),
               );
-          case "provider-thread.fork":
           case "terminal.cleanup":
+            return resourceCleanup.cleanupTerminals(effect.threadId).pipe(
+              Effect.mapError(
+                (cause) =>
+                  new OrchestrationEffectExecutionError({
+                    effectId: effect.id,
+                    effectType: effect.request.type,
+                    cause,
+                  }),
+              ),
+            );
           case "attachment.cleanup":
-            return Effect.fail(
+            return resourceCleanup.cleanupAttachments(effect.request.attachmentIds).pipe(
-              new OrchestrationEffectExecutionError({
+              Effect.mapError(
-                effectId: effect.id,
+                (cause) =>
-                effectType: effect.request.type,
+                  new OrchestrationEffectExecutionError({
-                cause: `No Shape 1 executor is registered for ${effect.request.type}.`,
+                    effectId: effect.id,
-              }),
+                    effectType: effect.request.type,
+                    cause,
+                  }),
+              ),
             );
         }
       },

apps/server/src/orchestration-v2/FoundationPersistence.test.ts+38−0

@@ -354,6 +354,44 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { · 353 unmodified lines
354	);	354	);
355	assert.equal(claims.filter(Option.isSome).length, 1);	355	assert.equal(claims.filter(Option.isSome).length, 1);
356	assert.equal(claims.filter(Option.isNone).length, 1);	356	assert.equal(claims.filter(Option.isNone).length, 1);
		357	const claimedByA = claims[0];
		358	const claimedByB = claims[1];
		359	if (Option.isSome(claimedByA)) {
		360	yield* outbox.succeed({ effectId: claimedByA.value.id, workerId: "worker-a" });
		361	}
		362	if (Option.isSome(claimedByB)) {
		363	yield* outbox.succeed({ effectId: claimedByB.value.id, workerId: "worker-b" });
		364	}
		365	}),
		366	);
		367
		368	it.effect("reclaims running effects immediately during startup recovery", () =>
		369	Effect.gen(function* () {
		370	const outbox = yield* EffectOutboxV2;
		371	const commandId = CommandId.make("command:foundation-reclaim-running");
		372	yield* outbox.enqueue([
		373	{
		374	id: "effect:foundation-reclaim-running",
		375	commandId,
		376	threadId: ThreadId.make("thread:foundation-reclaim-running"),
		377	request: {
		378	type: "provider-turn.start",
		379	runId: RunId.make("run:foundation-reclaim-running"),
		380	},
		381	},
		382	]);
		383	assert.isTrue(
		384	Option.isSome(
		385	yield* outbox.claimNext({ workerId: "crashed-worker", leaseDurationMs: 30_000 }),
		386	),
		387	);
		388	assert.equal(yield* outbox.reclaimRunning, 1);
		389	const reclaimed = yield* outbox.claimNext({
		390	workerId: "recovery-worker",
		391	leaseDurationMs: 30_000,
		392	});
		393	assert.isTrue(Option.isSome(reclaimed));
		394	if (Option.isSome(reclaimed)) assert.equal(reclaimed.value.attemptCount, 2);
357	}),	395	}),
358	);	396	);
359		397

apps/server/src/orchestration-v2/Orchestrator.ts+123−13

@@ -815,16 +815,91 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 814 unmodified lines
815	payload: updatedThread,	815	payload: updatedThread,
816	});	816	});
817		817
		818	if (command.type === "thread.delete") {
		819	const emitEvent = emit(events, command);
		820	const activeRunIds = new Set(
		821	projection.runs
		822	.filter((run) => ["queued", "starting", "running", "waiting"].includes(run.status))
		823	.map((run) => run.id),
		824	);
		825	for (const run of projection.runs.filter((candidate) => activeRunIds.has(candidate.id))) {
		826	yield* emitEvent({
		827	type: "run.updated",
		828	threadId: command.threadId,
		829	runId: run.id,
		830	providerInstanceId: run.providerInstanceId,
		831	occurredAt: now,
		832	payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now },
		833	});
		834	}
		835	for (const attempt of projection.attempts.filter(
		836	(candidate) =>
		837	activeRunIds.has(candidate.runId) &&
		838	(candidate.status === "pending" \|\| candidate.status === "running"),
		839	)) {
		840	const run = projection.runs.find((candidate) => candidate.id === attempt.runId)!;
		841	yield* emitEvent({
		842	type: "run-attempt.updated",
		843	threadId: command.threadId,
		844	runId: attempt.runId,
		845	nodeId: attempt.rootNodeId,
		846	providerInstanceId: run.providerInstanceId,
		847	occurredAt: now,
		848	payload: { ...attempt, status: "cancelled", completedAt: now },
		849	});
		850	}
		851	for (const node of projection.nodes.filter(
		852	(candidate) =>
		853	candidate.runId !== null &&
		854	activeRunIds.has(candidate.runId) &&
		855	["pending", "running", "waiting"].includes(candidate.status),
		856	)) {
		857	const run = projection.runs.find((candidate) => candidate.id === node.runId)!;
		858	yield* emitEvent({
		859	type: "node.updated",
		860	threadId: command.threadId,
		861	runId: run.id,
		862	nodeId: node.id,
		863	providerInstanceId: run.providerInstanceId,
		864	occurredAt: now,
		865	payload: { ...node, status: "cancelled", completedAt: now },
		866	});
		867	}
		868	for (const request of projection.runtimeRequests.filter(
		869	(candidate) => candidate.status === "pending",
		870	)) {
		871	yield* emitEvent({
		872	type: "runtime-request.updated",
		873	threadId: command.threadId,
		874	nodeId: request.nodeId,
		875	occurredAt: now,
		876	payload: {
		877	...request,
		878	status: "cancelled",
		879	responseCapability: {
		880	type: "not_resumable",
		881	reason: "The thread was deleted.",
		882	},
		883	resolvedAt: now,
		884	},
		885	});
		886	}
		887	}
		888
818	const detachSessionIds = new Set(	889	const detachSessionIds = new Set(
819	command.type === "thread.archive" \|\| command.type === "thread.delete"	890	command.type === "thread.archive" \|\| command.type === "thread.delete"
820	? projection.providerSessions.map((session) => session.id)	891	? projection.providerSessions.map((session) => session.id)
821	: command.type === "thread.runtime-mode.set"	892	: command.type === "thread.metadata.update" &&
822	? projection.providerSessions	893	command.worktreePath !== undefined &&
823	.filter(	894	command.worktreePath !== thread.worktreePath
824	(session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession,	895	? projection.providerSessions.map((session) => session.id)
825	)	896	: command.type === "thread.runtime-mode.set"
826	.map((session) => session.id)	897	? projection.providerSessions
827	: (providerSwitchPlan?.releaseProviderSessionIds ?? []),	898	.filter(
		899	(session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession,
		900	)
		901	.map((session) => session.id)
		902	: (providerSwitchPlan?.releaseProviderSessionIds ?? []),
828	);	903	);
829	if (detachSessionIds.size > 0) {	904	if (detachSessionIds.size > 0) {
830	const liveSessions = projection.providerSessions.filter(	905	const liveSessions = projection.providerSessions.filter(
@@ -854,9 +929,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 23 unmodified lines
854	? "Thread archived."	929	? "Thread archived."
855	: command.type === "thread.delete"	930	: command.type === "thread.delete"
856	? "Thread deleted."	931	? "Thread deleted."
857	: command.type === "thread.runtime-mode.set"	932	: command.type === "thread.metadata.update"
858	? "Runtime mode changed."	933	? "Workspace changed."
859	: "Provider or model selection changed.",	934	: command.type === "thread.runtime-mode.set"
		935	? "Runtime mode changed."
		936	: "Provider or model selection changed.",
860	},	937	},
861	});	938	});
862	const pendingEffect = {	939	const pendingEffect = {
@@ -871,9 +948,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 8 unmodified lines
871	? "Thread archived."	948	? "Thread archived."
872	: command.type === "thread.delete"	949	: command.type === "thread.delete"
873	? "Thread deleted."	950	? "Thread deleted."
874	: command.type === "thread.runtime-mode.set"	951	: command.type === "thread.metadata.update"
875	? "Runtime mode changed."	952	? "Workspace changed."
876	: "Provider or model selection changed.",	953	: command.type === "thread.runtime-mode.set"
		954	? "Runtime mode changed."
		955	: "Provider or model selection changed.",
877	},	956	},
878	} satisfies PendingOrchestrationEffectV2;	957	} satisfies PendingOrchestrationEffectV2;
879	yield* Ref.update(effects, (existing) => [...existing, pendingEffect]);	958	yield* Ref.update(effects, (existing) => [...existing, pendingEffect]);
@@ -881,6 +960,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 1 unmodified line
881	{ concurrency: 1, discard: true },	960	{ concurrency: 1, discard: true },
882	);	961	);
883	}	962	}
		963
		964	if (command.type === "thread.archive" \|\| command.type === "thread.delete") {
		965	yield* Ref.update(effects, (existing) => [
		966	...existing,
		967	{
		968	id: `effect:${command.commandId}:terminal.cleanup`,
		969	commandId: command.commandId,
		970	threadId: command.threadId,
		971	request: { type: "terminal.cleanup" },
		972	} satisfies PendingOrchestrationEffectV2,
		973	]);
		974	}
		975
		976	if (command.type === "thread.delete") {
		977	const attachmentIds = Array.from(
		978	new Set(
		979	projection.messages.flatMap((message) => message.attachments.map((item) => item.id)),
		980	),
		981	);
		982	if (attachmentIds.length > 0) {
		983	yield* Ref.update(effects, (existing) => [
		984	...existing,
		985	{
		986	id: `effect:${command.commandId}:attachment.cleanup`,
		987	commandId: command.commandId,
		988	threadId: command.threadId,
		989	request: { type: "attachment.cleanup", attachmentIds },
		990	} satisfies PendingOrchestrationEffectV2,
		991	]);
		992	}
		993	}
884	});	994	});
885		995
886	const dispatchProviderSessionDetach = Effect.fn("orchestrationV2.dispatch.providerSessionDetach")(	996	const dispatchProviderSessionDetach = Effect.fn("orchestrationV2.dispatch.providerSessionDetach")(

apps/server/src/orchestration-v2/ProviderAdapterRegistry.ts+37−1

@@ -1,5 +1,6 @@
1	import {	1	import {
2	ProviderInstanceId,	2	ProviderInstanceId,
		3	type OrchestrationV2ProviderCapabilities,
3	type ProviderDriverKind,	4	type ProviderDriverKind,
4	type ProviderInstanceConfig,	5	type ProviderInstanceConfig,
5	type ProviderInstanceConfigMap,	6	type ProviderInstanceConfigMap,
@@ -29,8 +30,14 @@ export class ProviderAdapterRegistryLookupError extends Schema.TaggedErrorClass< · 23 unmodified lines
29	}	30	}
30	}	31	}
31		32
		33	export class ProviderAdapterRegistryMetadataError extends Schema.TaggedErrorClass<ProviderAdapterRegistryMetadataError>()(
		34	"ProviderAdapterRegistryMetadataError",
		35	{ instanceId: ProviderInstanceId, cause: Schema.Defect() },
		36	) {}
		37
32	export const ProviderAdapterRegistryV2Error = Schema.Union([	38	export const ProviderAdapterRegistryV2Error = Schema.Union([
33	ProviderAdapterRegistryLookupError,	39	ProviderAdapterRegistryLookupError,
		40	ProviderAdapterRegistryMetadataError,
34	ProviderAdapterDriverCreateError,	41	ProviderAdapterDriverCreateError,
35	]);	42	]);
36	export type ProviderAdapterRegistryV2Error = typeof ProviderAdapterRegistryV2Error.Type;	43	export type ProviderAdapterRegistryV2Error = typeof ProviderAdapterRegistryV2Error.Type;
@@ -39,7 +46,16 @@ export interface ProviderAdapterRegistryV2Shape { · 2 unmodified lines
39	readonly get: (	46	readonly get: (
40	provider: ProviderKind,	47	provider: ProviderKind,
41	) => Effect.Effect<ProviderAdapterV2Shape, ProviderAdapterRegistryV2Error>;	48	) => Effect.Effect<ProviderAdapterV2Shape, ProviderAdapterRegistryV2Error>;
42	readonly list: () => Effect.Effect<ReadonlyArray<ProviderKind>>;	49	readonly list: () => Effect.Effect<ReadonlyArray<ProviderInstanceId>>;
		50	readonly getMetadata?: (instanceId: ProviderInstanceId) => Effect.Effect<
		51	{
		52	readonly driver: ProviderAdapterV2Shape["driver"];
		53	readonly continuationKey: string;
		54	readonly enabled: boolean;
		55	readonly capabilities: OrchestrationV2ProviderCapabilities;
		56	},
		57	ProviderAdapterRegistryV2Error
		58	>;
43	}	59	}
44		60
45	export class ProviderAdapterRegistryV2 extends Context.Service<	61	export class ProviderAdapterRegistryV2 extends Context.Service<
@@ -75,6 +91,26 @@ export const layerFromProviderInstanceRegistry: Layer.Layer< · 29 unmodified lines
75	instances.listInstances.pipe(	91	instances.listInstances.pipe(
76	Effect.map((available) => available.map((instance) => instance.instanceId)),	92	Effect.map((available) => available.map((instance) => instance.instanceId)),
77	),	93	),
		94	getMetadata: (instanceId) =>
		95	Effect.gen(function* () {
		96	const instance = yield* instances.getInstance(instanceId);
		97	if (instance === undefined) {
		98	return yield* new ProviderAdapterRegistryLookupError({ instanceId });
		99	}
		100	const capabilities = yield* instance.orchestrationAdapter
		101	.getCapabilities()
		102	.pipe(
		103	Effect.mapError(
		104	(cause) => new ProviderAdapterRegistryMetadataError({ instanceId, cause }),
		105	),
		106	);
		107	return {
		108	driver: instance.driverKind,
		109	continuationKey: instance.continuationIdentity.continuationKey,
		110	enabled: instance.enabled,
		111	capabilities,
		112	};
		113	}),
78	});	114	});
79	}),	115	}),
80	);	116	);

apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts+260−0

@@ -0,0 +1,260 @@
	1	import { assert, it, vi } from "@effect/vitest";
	2	import {
	3	NodeId,
	4	RuntimeRequestId,
	5	ThreadId,
	6	type OrchestrationV2ThreadProjection,
	7	} from "@t3tools/contracts";
	8	import * as Effect from "effect/Effect";
	9	import * as Layer from "effect/Layer";
	10	import * as Ref from "effect/Ref";
	11
	12	import * as EffectWorker from "./EffectWorker.ts";
	13	import * as EffectOutbox from "./EffectOutbox.ts";
	14	import * as EventSink from "./EventSink.ts";
	15	import * as IdAllocator from "./IdAllocator.ts";
	16	import * as ProjectionStore from "./ProjectionStore.ts";
	17	import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
	18	import * as ProviderSessionManager from "./ProviderSessionManager.ts";
	19
	20	const { decideProviderRuntimeRecovery } = ProviderRuntimeRecovery;
	21
	22	it("retries process and transport failures only within the idempotent retry budget", () => {
	23	assert.deepEqual(
	24	decideProviderRuntimeRecovery({
	25	kind: "process_exited",
	26	attempt: 1,
	27	maxAttempts: 3,
	28	idempotent: true,
	29	online: true,
	30	}),
	31	{ type: "retry_now" },
	32	);
	33	assert.equal(
	34	decideProviderRuntimeRecovery({
	35	kind: "transport_unavailable",
	36	attempt: 3,
	37	maxAttempts: 3,
	38	idempotent: true,
	39	online: true,
	40	}).type,
	41	"terminalize",
	42	);
	43	});
	44
	45	it("waits for connectivity and requires retry-after for rate limits", () => {
	46	assert.deepEqual(
	47	decideProviderRuntimeRecovery({
	48	kind: "network_unavailable",
	49	attempt: 0,
	50	maxAttempts: 3,
	51	idempotent: true,
	52	online: false,
	53	}),
	54	{ type: "wait_for_connectivity" },
	55	);
	56	assert.deepEqual(
	57	decideProviderRuntimeRecovery({
	58	kind: "provider_rate_limited",
	59	attempt: 0,
	60	maxAttempts: 3,
	61	idempotent: true,
	62	retryAfterMs: 250,
	63	online: true,
	64	}),
	65	{ type: "retry_after", delayMs: 250 },
	66	);
	67	});
	68
	69	it("terminalizes non-recoverable provider failures", () => {
	70	for (const kind of [
	71	"provider_quota_exceeded",
	72	"auth_invalid",
	73	"permission_denied",
	74	"invalid_request",
	75	"unsupported_model",
	76	] as const) {
	77	assert.equal(
	78	decideProviderRuntimeRecovery({
	79	kind,
	80	attempt: 0,
	81	maxAttempts: 3,
	82	idempotent: true,
	83	online: true,
	84	}).type,
	85	"terminalize",
	86	);
	87	}
	88	});
	89
	90	it("classifies wrapped provider failures without provider-name checks", () => {
	91	assert.deepEqual(
	92	ProviderRuntimeRecovery.classifyProviderRuntimeFailure({
	93	_tag: "ProviderSessionOpenError",
	94	cause: { code: "rate_limit_exceeded", retryAfterMs: 125 },
	95	}),
	96	{ kind: "provider_rate_limited", retryAfterMs: 125 },
	97	);
	98	assert.deepEqual(
	99	ProviderRuntimeRecovery.classifyProviderRuntimeFailure({
	100	cause: { status: 401, message: "authentication failed" },
	101	}),
	102	{ kind: "auth_invalid" },
	103	);
	104	});
	105
	106	it.effect("executes bounded transport retries and returns the resumed value", () =>
	107	Effect.gen(function* () {
	108	const attempts = yield* Ref.make(0);
	109	const value = yield* ProviderRuntimeRecovery.recoverWithPolicy({
	110	operation: Ref.getAndUpdate(attempts, (count) => count + 1).pipe(
	111	Effect.flatMap((count) =>
	112	count < 2 ? Effect.fail("transport down") : Effect.succeed("resumed"),
	113	),
	114	),
	115	classify: () => ({ kind: "transport_unavailable" }),
	116	connectivity: { isOnline: Effect.succeed(true), awaitOnline: Effect.void },
	117	maxAttempts: 3,
	118	idempotent: true,
	119	});
	120	assert.equal(value, "resumed");
	121	assert.equal(yield* Ref.get(attempts), 3);
	122	}),
	123	);
	124
	125	it.effect("waits for connectivity before retrying a network failure", () =>
	126	Effect.gen(function* () {
	127	const attempts = yield* Ref.make(0);
	128	const online = yield* Ref.make(false);
	129	const waits = yield* Ref.make(0);
	130	const value = yield* ProviderRuntimeRecovery.recoverWithPolicy({
	131	operation: Ref.getAndUpdate(attempts, (count) => count + 1).pipe(
	132	Effect.flatMap((count) =>
	133	count === 0 ? Effect.fail("offline") : Effect.succeed("resumed"),
	134	),
	135	),
	136	classify: () => ({ kind: "network_unavailable" }),
	137	connectivity: {
	138	isOnline: Ref.get(online),
	139	awaitOnline: Ref.set(online, true).pipe(
	140	Effect.andThen(Ref.update(waits, (count) => count + 1)),
	141	),
	142	},
	143	maxAttempts: 3,
	144	idempotent: true,
	145	});
	146	assert.equal(value, "resumed");
	147	assert.equal(yield* Ref.get(waits), 1);
	148	}),
	149	);
	150
	151	it.effect("does not retry unrecoverable failures", () =>
	152	Effect.gen(function* () {
	153	const attempts = yield* Ref.make(0);
	154	const result = yield* Effect.result(
	155	ProviderRuntimeRecovery.recoverWithPolicy({
	156	operation: Ref.update(attempts, (count) => count + 1).pipe(
	157	Effect.andThen(Effect.fail("invalid credentials")),
	158	),
	159	classify: () => ({ kind: "auth_invalid" }),
	160	connectivity: { isOnline: Effect.succeed(true), awaitOnline: Effect.void },
	161	maxAttempts: 3,
	162	idempotent: true,
	163	}),
	164	);
	165	assert.equal(result._tag, "Failure");
	166	assert.equal(yield* Ref.get(attempts), 1);
	167	}),
	168	);
	169
	170	it.effect("drains durable effects before reporting recovery complete", () =>
	171	Effect.gen(function* () {
	172	const runs = yield* Ref.make(0);
	173	const layer = ProviderRuntimeRecovery.layer.pipe(
	174	Layer.provide(
	175	Layer.mergeAll(
	176	Layer.mock(ProjectionStore.ProjectionStoreV2)({
	177	getShellSnapshot: () =>
	178	Effect.succeed({
	179	schemaVersion: 2,
	180	snapshotSequence: 0,
	181	threads: [],
	182	archivedThreads: [],
	183	}),
	184	}),
	185	Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}),
	186	Layer.mock(EventSink.EventSinkV2)({}),
	187	IdAllocator.layer,
	188	Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
	189	runOnce: Ref.getAndUpdate(runs, (count) => count + 1).pipe(
	190	Effect.map((count) => count < 2),
	191	),
	192	}),
	193	Layer.mock(EffectOutbox.EffectOutboxV2)({ reclaimRunning: Effect.succeed(0) }),
	194	),
	195	),
	196	);
	197	const summary = yield* Effect.gen(function* () {
	198	return yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).recover;
	199	}).pipe(Effect.provide(layer));
	200	assert.deepEqual(summary, { resumedSessions: 0, terminalizedRuns: 0, executedEffects: 2 });
	201	}),
	202	);
	203
	204	it.effect("expires orphaned runtime requests before command readiness", () => {
	205	const threadId = ThreadId.make("thread_recovery_requests");
	206	let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] \| null =
	207	null;
	208	const committed = vi.fn(
	209	(input: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0]) => {
	210	committedInput = input;
	211	return Effect.succeed({ committed: true } as never);
	212	},
	213	);
	214	const projection = {
	215	thread: { id: threadId },
	216	runtimeRequests: [
	217	{
	218	id: RuntimeRequestId.make("request_orphaned"),
	219	nodeId: NodeId.make("node_orphaned"),
	220	status: "pending",
	221	responseCapability: { type: "not_resumable", reason: "old process" },
	222	},
	223	],
	224	providerSessions: [],
	225	runs: [],
	226	nodes: [],
	227	} as unknown as OrchestrationV2ThreadProjection;
	228	const layer = ProviderRuntimeRecovery.layer.pipe(
	229	Layer.provide(
	230	Layer.mergeAll(
	231	Layer.mock(ProjectionStore.ProjectionStoreV2)({
	232	getShellSnapshot: () =>
	233	Effect.succeed({
	234	schemaVersion: 2,
	235	snapshotSequence: 0,
	236	threads: [{ id: threadId }],
	237	archivedThreads: [],
	238	} as never),
	239	getThreadProjection: () => Effect.succeed(projection),
	240	}),
	241	Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}),
	242	Layer.mock(EventSink.EventSinkV2)({ commitCommand: committed }),
	243	IdAllocator.layer,
	244	Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({ runOnce: Effect.succeed(false) }),
	245	Layer.mock(EffectOutbox.EffectOutboxV2)({ reclaimRunning: Effect.succeed(0) }),
	246	),
	247	),
	248	);
	249	return Effect.gen(function* () {
	250	yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).recover;
	251	const command = committedInput;
	252	assert.isNotNull(command);
	253	if (command === null) return;
	254	assert.equal(command?.events[0]?.type, "runtime-request.updated");
	255	if (command?.events[0]?.type === "runtime-request.updated") {
	256	assert.equal(command.events[0].payload.status, "expired");
	257	assert.equal(command.events[0].payload.responseCapability.type, "not_resumable");
	258	}
	259	}).pipe(Effect.provide(layer));
	260	});

apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts+508−0

@@ -0,0 +1,508 @@
	1	import {
	2	CommandId,
	3	type OrchestrationV2DomainEvent,
	4	type OrchestrationV2ThreadProjection,
	5	ThreadId,
	6	} from "@t3tools/contracts";
	7	import * as Context from "effect/Context";
	8	import * as DateTime from "effect/DateTime";
	9	import * as Duration from "effect/Duration";
	10	import * as Effect from "effect/Effect";
	11	import * as Layer from "effect/Layer";
	12	import * as Schema from "effect/Schema";
	13
	14	import * as EffectWorker from "./EffectWorker.ts";
	15	import * as EffectOutbox from "./EffectOutbox.ts";
	16	import * as EventSink from "./EventSink.ts";
	17	import * as IdAllocator from "./IdAllocator.ts";
	18	import * as ProjectionStore from "./ProjectionStore.ts";
	19	import * as ProviderSessionManager from "./ProviderSessionManager.ts";
	20
	21	export class ProviderRuntimeRecoveryError extends Schema.TaggedErrorClass<ProviderRuntimeRecoveryError>()(
	22	"ProviderRuntimeRecoveryError",
	23	{
	24	operation: Schema.Literals([
	25	"read-projections",
	26	"resume-session",
	27	"terminalize",
	28	"drain-outbox",
	29	]),
	30	threadId: Schema.optional(ThreadId),
	31	cause: Schema.Defect(),
	32	},
	33	) {
	34	override get message(): string {
	35	return `Provider runtime recovery failed during ${this.operation}.`;
	36	}
	37	}
	38
	39	export interface ProviderRuntimeRecoverySummary {
	40	readonly resumedSessions: number;
	41	readonly terminalizedRuns: number;
	42	readonly executedEffects: number;
	43	}
	44
	45	export const ProviderRuntimeFailureKind = Schema.Literals([
	46	"process_exited",
	47	"transport_unavailable",
	48	"network_unavailable",
	49	"provider_rate_limited",
	50	"provider_quota_exceeded",
	51	"auth_invalid",
	52	"permission_denied",
	53	"invalid_request",
	54	"unsupported_model",
	55	]);
	56	export type ProviderRuntimeFailureKind = typeof ProviderRuntimeFailureKind.Type;
	57
	58	export type ProviderRuntimeRecoveryDecision =
	59	\| { readonly type: "retry_now" }
	60	\| { readonly type: "retry_after"; readonly delayMs: number }
	61	\| { readonly type: "wait_for_connectivity" }
	62	\| { readonly type: "terminalize"; readonly reason: string };
	63
	64	export function decideProviderRuntimeRecovery(input: {
	65	readonly kind: ProviderRuntimeFailureKind;
	66	readonly attempt: number;
	67	readonly maxAttempts: number;
	68	readonly idempotent: boolean;
	69	readonly retryAfterMs?: number;
	70	readonly online: boolean;
	71	}): ProviderRuntimeRecoveryDecision {
	72	if (input.attempt >= input.maxAttempts) {
	73	return { type: "terminalize", reason: "Provider recovery retry budget was exhausted." };
	74	}
	75	switch (input.kind) {
	76	case "process_exited":
	77	case "transport_unavailable":
	78	return input.idempotent
	79	? { type: "retry_now" }
	80	: { type: "terminalize", reason: "The interrupted operation is not idempotent." };
	81	case "network_unavailable":
	82	return input.online ? { type: "retry_now" } : { type: "wait_for_connectivity" };
	83	case "provider_rate_limited":
	84	return input.idempotent && input.retryAfterMs !== undefined
	85	? { type: "retry_after", delayMs: Math.max(0, input.retryAfterMs) }
	86	: {
	87	type: "terminalize",
	88	reason: "Rate-limit recovery requires retry-after and an idempotent operation.",
	89	};
	90	case "provider_quota_exceeded":
	91	case "auth_invalid":
	92	case "permission_denied":
	93	case "invalid_request":
	94	case "unsupported_model":
	95	return { type: "terminalize", reason: `Provider failure ${input.kind} is not recoverable.` };
	96	}
	97	}
	98
	99	export interface ConnectivityServiceShape {
	100	readonly isOnline: Effect.Effect<boolean>;
	101	readonly awaitOnline: Effect.Effect<void>;
	102	}
	103
	104	export class ConnectivityService extends Context.Reference<ConnectivityServiceShape>(
	105	"t3/orchestration-v2/ConnectivityService",
	106	{
	107	defaultValue: () => ({ isOnline: Effect.succeed(true), awaitOnline: Effect.void }),
	108	},
	109	) {}
	110
	111	export interface ClassifiedProviderRuntimeFailure {
	112	readonly kind: ProviderRuntimeFailureKind;
	113	readonly retryAfterMs?: number;
	114	}
	115
	116	export function classifyProviderRuntimeFailure(cause: unknown): ClassifiedProviderRuntimeFailure {
	117	const record =
	118	cause !== null && typeof cause === "object"
	119	? (cause as Record<string, unknown>)
	120	: ({} as Record<string, unknown>);
	121	const serialized = (() => {
	122	try {
	123	return JSON.stringify(cause);
	124	} catch {
	125	return "";
	126	}
	127	})();
	128	const description = [
	129	record._tag,
	130	record.code,
	131	record.status,
	132	record.message,
	133	record.detail,
	134	serialized,
	135	]
	136	.filter((value) => typeof value === "string" \|\| typeof value === "number")
	137	.join(" ")
	138	.toLowerCase();
	139	const nestedCause =
	140	record.cause !== null && typeof record.cause === "object"
	141	? (record.cause as Record<string, unknown>)
	142	: undefined;
	143	const retryAfter =
	144	record.retryAfterMs ??
	145	record.retry_after_ms ??
	146	nestedCause?.retryAfterMs ??
	147	nestedCause?.retry_after_ms;
	148	const retryAfterMs = typeof retryAfter === "number" ? retryAfter : undefined;
	149	if (description.includes("rate") && description.includes("limit")) {
	150	return {
	151	kind: "provider_rate_limited",
	152	...(retryAfterMs === undefined ? {} : { retryAfterMs }),
	153	};
	154	}
	155	if (description.includes("quota")) return { kind: "provider_quota_exceeded" };
	156	if (description.includes("auth") \|\| description.includes("unauthorized")) {
	157	return { kind: "auth_invalid" };
	158	}
	159	if (description.includes("permission") \|\| description.includes("forbidden")) {
	160	return { kind: "permission_denied" };
	161	}
	162	if (description.includes("unsupported") && description.includes("model")) {
	163	return { kind: "unsupported_model" };
	164	}
	165	if (description.includes("invalid") && description.includes("request")) {
	166	return { kind: "invalid_request" };
	167	}
	168	if (
	169	description.includes("network") \|\|
	170	description.includes("offline") \|\|
	171	description.includes("econnreset") \|\|
	172	description.includes("enotfound")
	173	) {
	174	return { kind: "network_unavailable" };
	175	}
	176	if (description.includes("exit") \|\| description.includes("terminated")) {
	177	return { kind: "process_exited" };
	178	}
	179	return { kind: "transport_unavailable" };
	180	}
	181
	182	export class ProviderRuntimeFailureClassifier extends Context.Reference<{
	183	readonly classify: (cause: unknown) => ClassifiedProviderRuntimeFailure;
	184	}>("t3/orchestration-v2/ProviderRuntimeFailureClassifier", {
	185	defaultValue: () => ({ classify: classifyProviderRuntimeFailure }),
	186	}) {}
	187
	188	export class ProviderRuntimeRecoveryPolicy extends Context.Reference<{
	189	readonly maxAttempts: number;
	190	}>("t3/orchestration-v2/ProviderRuntimeRecoveryPolicy", {
	191	defaultValue: () => ({ maxAttempts: 3 }),
	192	}) {}
	193
	194	export function recoverWithPolicy<A, E>(input: {
	195	readonly operation: Effect.Effect<A, E>;
	196	readonly classify: (cause: E) => ClassifiedProviderRuntimeFailure;
	197	readonly connectivity: ConnectivityServiceShape;
	198	readonly maxAttempts: number;
	199	readonly idempotent: boolean;
	200	}): Effect.Effect<A, E> {
	201	const attempt = (attemptNumber: number): Effect.Effect<A, E> =>
	202	input.operation.pipe(
	203	Effect.catch((cause) =>
	204	Effect.gen(function* () {
	205	const classified = input.classify(cause);
	206	const online = yield* input.connectivity.isOnline;
	207	const decision = decideProviderRuntimeRecovery({
	208	kind: classified.kind,
	209	attempt: attemptNumber,
	210	maxAttempts: input.maxAttempts,
	211	idempotent: input.idempotent,
	212	online,
	213	...(classified.retryAfterMs === undefined
	214	? {}
	215	: { retryAfterMs: classified.retryAfterMs }),
	216	});
	217	switch (decision.type) {
	218	case "retry_now":
	219	return yield* attempt(attemptNumber + 1);
	220	case "retry_after":
	221	yield* Effect.sleep(Duration.millis(decision.delayMs));
	222	return yield* attempt(attemptNumber + 1);
	223	case "wait_for_connectivity":
	224	yield* input.connectivity.awaitOnline;
	225	return yield* attempt(attemptNumber + 1);
	226	case "terminalize":
	227	return yield* Effect.fail(cause);
	228	}
	229	}),
	230	),
	231	);
	232	return attempt(1);
	233	}
	234
	235	export class ProviderRuntimeRecoveryService extends Context.Service<
	236	ProviderRuntimeRecoveryService,
	237	{
	238	readonly recover: Effect.Effect<ProviderRuntimeRecoverySummary, ProviderRuntimeRecoveryError>;
	239	}
	240	>()("t3/orchestration-v2/ProviderRuntimeRecoveryService") {}
	241
	242	function activeRuns(projection: OrchestrationV2ThreadProjection) {
	243	return projection.runs.filter(
	244	(run) => run.status === "starting" \|\| run.status === "running" \|\| run.status === "waiting",
	245	);
	246	}
	247
	248	export const make = Effect.gen(function* () {
	249	const projections = yield* ProjectionStore.ProjectionStoreV2;
	250	const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
	251	const eventSink = yield* EventSink.EventSinkV2;
	252	const ids = yield* IdAllocator.IdAllocatorV2;
	253	const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
	254	const outbox = yield* EffectOutbox.EffectOutboxV2;
	255	const connectivity = yield* ConnectivityService;
	256	const classifier = yield* ProviderRuntimeFailureClassifier;
	257	const recoveryPolicy = yield* ProviderRuntimeRecoveryPolicy;
	258
	259	const terminalize = Effect.fn("ProviderRuntimeRecoveryService.terminalize")(function* (
	260	projection: OrchestrationV2ThreadProjection,
	261	detail: string,
	262	) {
	263	const now = yield* DateTime.now;
	264	const runs = activeRuns(projection);
	265	if (runs.length === 0) return 0;
	266	const commandId = CommandId.make(`command:recovery:terminalize:${projection.thread.id}`);
	267	const allocateEventId = () =>
	268	ids.allocate.event({ threadId: projection.thread.id, commandId }).pipe(
	269	Effect.mapError(
	270	(cause) =>
	271	new ProviderRuntimeRecoveryError({
	272	operation: "terminalize",
	273	threadId: projection.thread.id,
	274	cause,
	275	}),
	276	),
	277	);
	278	const events: Array<OrchestrationV2DomainEvent> = [];
	279	for (const run of runs) {
	280	events.push({
	281	id: yield* allocateEventId(),
	282	type: "run.updated",
	283	threadId: projection.thread.id,
	284	runId: run.id,
	285	providerInstanceId: run.providerInstanceId,
	286	occurredAt: now,
	287	payload: { ...run, status: "failed", completedAt: now },
	288	});
	289	for (const node of projection.nodes.filter(
	290	(candidate) =>
	291	candidate.runId === run.id &&
	292	(candidate.status === "pending" \|\|
	293	candidate.status === "running" \|\|
	294	candidate.status === "waiting"),
	295	)) {
	296	events.push({
	297	id: yield* allocateEventId(),
	298	type: "node.updated",
	299	threadId: projection.thread.id,
	300	runId: run.id,
	301	nodeId: node.id,
	302	providerInstanceId: run.providerInstanceId,
	303	occurredAt: now,
	304	payload: { ...node, status: "failed", completedAt: now },
	305	});
	306	}
	307	}
	308	yield* eventSink
	309	.commitCommand({
	310	commandId,
	311	threadId: projection.thread.id,
	312	commandType: "provider-runtime.recovery-terminalize",
	313	acceptedAt: now,
	314	events,
	315	effects: [],
	316	})
	317	.pipe(
	318	Effect.mapError(
	319	(cause) =>
	320	new ProviderRuntimeRecoveryError({
	321	operation: "terminalize",
	322	threadId: projection.thread.id,
	323	cause: { detail, cause },
	324	}),
	325	),
	326	);
	327	return runs.length;
	328	});
	329
	330	const expireOrphanedRequests = Effect.fn("ProviderRuntimeRecoveryService.expireOrphanedRequests")(
	331	function* (projection: OrchestrationV2ThreadProjection) {
	332	const requests = projection.runtimeRequests.filter((request) => request.status === "pending");
	333	if (requests.length === 0) return;
	334	const now = yield* DateTime.now;
	335	const commandId = CommandId.make(`command:recovery:expire-requests:${projection.thread.id}`);
	336	const events: Array<OrchestrationV2DomainEvent> = [];
	337	for (const request of requests) {
	338	events.push({
	339	id: yield* ids.allocate.event({ threadId: projection.thread.id, commandId }).pipe(
	340	Effect.mapError(
	341	(cause) =>
	342	new ProviderRuntimeRecoveryError({
	343	operation: "terminalize",
	344	threadId: projection.thread.id,
	345	cause,
	346	}),
	347	),
	348	),
	349	type: "runtime-request.updated",
	350	threadId: projection.thread.id,
	351	nodeId: request.nodeId,
	352	occurredAt: now,
	353	payload: {
	354	...request,
	355	status: "expired",
	356	responseCapability: {
	357	type: "not_resumable",
	358	reason: "The server restarted before this runtime request was resolved.",
	359	},
	360	resolvedAt: now,
	361	},
	362	});
	363	}
	364	yield* eventSink
	365	.commitCommand({
	366	commandId,
	367	threadId: projection.thread.id,
	368	commandType: "provider-runtime.expire-orphaned-requests",
	369	acceptedAt: now,
	370	events,
	371	effects: [],
	372	})
	373	.pipe(
	374	Effect.mapError(
	375	(cause) =>
	376	new ProviderRuntimeRecoveryError({
	377	operation: "terminalize",
	378	threadId: projection.thread.id,
	379	cause,
	380	}),
	381	),
	382	);
	383	},
	384	);
	385
	386	const recoverProjection = Effect.fn("ProviderRuntimeRecoveryService.recoverProjection")(
	387	function* (projection: OrchestrationV2ThreadProjection) {
	388	yield* expireOrphanedRequests(projection);
	389	let resumedSessions = 0;
	390	for (const session of projection.providerSessions.filter(
	391	(candidate) => candidate.status !== "stopped" && candidate.status !== "error",
	392	)) {
	393	const resume = sessions
	394	.open({
	395	threadId: projection.thread.id,
	396	providerSessionId: session.id,
	397	modelSelection: projection.thread.modelSelection,
	398	runtimePolicy: {
	399	runtimeMode: projection.thread.runtimeMode,
	400	interactionMode: projection.thread.interactionMode,
	401	cwd: session.cwd,
	402	},
	403	resumeFromSession: session,
	404	})
	405	.pipe(
	406	Effect.flatMap((runtime) =>
	407	Effect.forEach(
	408	projection.providerThreads.filter(
	409	(providerThread) =>
	410	providerThread.providerSessionId === session.id &&
	411	providerThread.appThreadId === projection.thread.id &&
	412	providerThread.status !== "closed" &&
	413	providerThread.status !== "archived",
	414	),
	415	(providerThread) =>
	416	runtime.resumeThread({
	417	providerThread,
	418	threadId: projection.thread.id,
	419	modelSelection: projection.thread.modelSelection,
	420	runtimePolicy: {
	421	runtimeMode: projection.thread.runtimeMode,
	422	interactionMode: projection.thread.interactionMode,
	423	cwd: session.cwd,
	424	},
	425	}),
	426	{ discard: true },
	427	),
	428	),
	429	);
	430	const recovered = yield* Effect.result(
	431	recoverWithPolicy({
	432	operation: resume,
	433	classify: classifier.classify,
	434	connectivity,
	435	maxAttempts: recoveryPolicy.maxAttempts,
	436	idempotent: true,
	437	}),
	438	);
	439	if (recovered._tag === "Failure") {
	440	const terminalizedRuns = yield* terminalize(
	441	projection,
	442	`Unable to resume provider session ${session.id}.`,
	443	);
	444	return { resumedSessions, terminalizedRuns };
	445	}
	446	resumedSessions += 1;
	447	}
	448	if (
	449	resumedSessions === 0 &&
	450	projection.runs.some((run) => run.status === "running" \|\| run.status === "waiting")
	451	) {
	452	const terminalizedRuns = yield* terminalize(
	453	projection,
	454	"No resumable provider session remained for the active run.",
	455	);
	456	return { resumedSessions, terminalizedRuns };
	457	}
	458	return { resumedSessions, terminalizedRuns: 0 };
	459	},
	460	);
	461
	462	const recover = Effect.gen(function* () {
	463	const shell = yield* projections
	464	.getShellSnapshot()
	465	.pipe(
	466	Effect.mapError(
	467	(cause) => new ProviderRuntimeRecoveryError({ operation: "read-projections", cause }),
	468	),
	469	);
	470	let resumedSessions = 0;
	471	let terminalizedRuns = 0;
	472	for (const thread of shell.threads) {
	473	const projection = yield* projections.getThreadProjection(thread.id).pipe(
	474	Effect.mapError(
	475	(cause) =>
	476	new ProviderRuntimeRecoveryError({
	477	operation: "read-projections",
	478	threadId: thread.id,
	479	cause,
	480	}),
	481	),
	482	);
	483	const result = yield* recoverProjection(projection);
	484	resumedSessions += result.resumedSessions;
	485	terminalizedRuns += result.terminalizedRuns;
	486	}
	487	let executedEffects = 0;
	488	yield* outbox.reclaimRunning.pipe(
	489	Effect.mapError(
	490	(cause) => new ProviderRuntimeRecoveryError({ operation: "drain-outbox", cause }),
	491	),
	492	);
	493	while (
	494	yield* worker.runOnce.pipe(
	495	Effect.mapError(
	496	(cause) => new ProviderRuntimeRecoveryError({ operation: "drain-outbox", cause }),
	497	),
	498	)
	499	) {
	500	executedEffects += 1;
	501	}
	502	return { resumedSessions, terminalizedRuns, executedEffects };
	503	});
	504
	505	return ProviderRuntimeRecoveryService.of({ recover });
	506	});
	507
	508	export const layer = Layer.effect(ProviderRuntimeRecoveryService, make);

apps/server/src/orchestration-v2/ProviderSessionManager.test.ts+0−20

@@ -1205,23 +1205,3 @@ it.effect( · 1204 unmodified lines
1205	);	1205	);
1206	}),	1206	}),
1207	);	1207	);
1208
1209	it.todo(
1210	"ProviderRuntimeRecoveryService restarts process_exited or transport_unavailable sessions with a bounded retry policy and resumes the native provider thread",
1211	);
1212
1213	it.todo(
1214	"ProviderRuntimeRecoveryService waits for ConnectivityService online before recovering network_unavailable sessions",
1215	);
1216
1217	it.todo(
1218	"ProviderRuntimeRecoveryService does not auto-recover provider_quota_exceeded, auth_invalid, permission_denied, invalid_request, or unsupported_model failures",
1219	);
1220
1221	it.todo(
1222	"ProviderRuntimeRecoveryService retries provider_rate_limited failures only when retry-after, idempotency, and retry budget allow it",
1223	);
1224
1225	it.todo(
1226	"ProviderRuntimeRecoveryService marks pending approvals and user-input requests non-live when a provider session is released or crashes",
1227	);

apps/server/src/orchestration-v2/ProviderSessionTransitionPolicy.test.ts+20−0

@@ -88,3 +88,23 @@ it("uses portable handoff for incompatible continuation identities", () => { · 87 unmodified lines
88	{ type: "create_with_handoff" },	88	{ type: "create_with_handoff" },
89	);	89	);
90	});	90	});
		91
		92	it("uses portable handoff for cross-driver transitions", () => {
		93	const claudeDriver = ProviderDriverKind.make("claude");
		94	assert.deepEqual(
		95	decideProviderSessionTransition({
		96	current: base,
		97	target: {
		98	...base,
		99	driver: claudeDriver,
		100	continuationIdentity: { driverKind: claudeDriver, continuationKey: "claude:account:one" },
		101	modelSelection: {
		102	instanceId: ProviderInstanceId.make("claude"),
		103	model: "claude-opus-4-1",
		104	},
		105	available: true,
		106	},
		107	}),
		108	{ type: "create_with_handoff" },
		109	);
		110	});

apps/server/src/orchestration-v2/ProviderSwitchService.test.ts+113−0

@@ -0,0 +1,113 @@
	1	import { assert, it } from "@effect/vitest";
	2	import {
	3	ProviderDriverKind,
	4	ProviderInstanceId,
	5	ProviderSessionId,
	6	ThreadId,
	7	type OrchestrationV2ThreadProjection,
	8	} from "@t3tools/contracts";
	9	import * as DateTime from "effect/DateTime";
	10	import * as Effect from "effect/Effect";
	11	import * as Layer from "effect/Layer";
	12
	13	import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
	14	import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
	15	import * as ProviderSwitch from "./ProviderSwitchService.ts";
	16
	17	const driver = ProviderDriverKind.make("codex");
	18	const currentInstanceId = ProviderInstanceId.make("codex_primary");
	19	const currentSessionId = ProviderSessionId.make("session_primary");
	20	const now = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
	21	const capabilitiesWithoutModelSwitch = {
	22	...CodexProviderCapabilitiesV2,
	23	sessions: {
	24	...CodexProviderCapabilitiesV2.sessions,
	25	supportsModelSwitchInSession: false,
	26	},
	27	};
	28
	29	function projection(): OrchestrationV2ThreadProjection {
	30	return {
	31	thread: {
	32	id: ThreadId.make("thread_switch_service"),
	33	modelSelection: { instanceId: currentInstanceId, model: "gpt-5.1-codex" },
	34	runtimeMode: "full-access",
	35	interactionMode: "default",
	36	worktreePath: "/repo",
	37	},
	38	providerSessions: [
	39	{
	40	id: currentSessionId,
	41	providerInstanceId: currentInstanceId,
	42	status: "ready",
	43	cwd: "/repo",
	44	capabilities: capabilitiesWithoutModelSwitch,
	45	updatedAt: now,
	46	},
	47	],
	48	providerThreads: [],
	49	} as unknown as OrchestrationV2ThreadProjection;
	50	}
	51
	52	function testLayer(metadata: Readonly<Record<string, { continuationKey: string }>>) {
	53	const registry = Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
	54	getMetadata: (instanceId) => {
	55	const value = metadata[instanceId];
	56	return value === undefined
	57	? Effect.fail(
	58	new ProviderAdapterRegistry.ProviderAdapterRegistryLookupError({ instanceId }),
	59	)
	60	: Effect.succeed({
	61	driver,
	62	continuationKey: value.continuationKey,
	63	enabled: true,
	64	capabilities: capabilitiesWithoutModelSwitch,
	65	});
	66	},
	67	});
	68	return ProviderSwitch.layer.pipe(Layer.provide(registry));
	69	}
	70
	71	it.effect(
	72	"restarts and releases the current session for unsupported in-session model changes",
	73	() =>
	74	Effect.gen(function* () {
	75	const service = yield* ProviderSwitch.ProviderSwitchServiceV2;
	76	const result = yield* service.plan({
	77	projection: projection(),
	78	targetModelSelection: { instanceId: currentInstanceId, model: "gpt-5.2-codex" },
	79	});
	80	assert.equal(result.transition.type, "restart_and_resume");
	81	assert.deepEqual(result.releaseProviderSessionIds, [currentSessionId]);
	82	}).pipe(
	83	Effect.provide(
	84	testLayer({ [currentInstanceId]: { continuationKey: "codex:account:primary" } }),
	85	),
	86	),
	87	);
	88
	89	it.effect("distinguishes compatible and incompatible instances of the same driver", () =>
	90	Effect.gen(function* () {
	91	const service = yield* ProviderSwitch.ProviderSwitchServiceV2;
	92	const compatibleId = ProviderInstanceId.make("codex_compatible");
	93	const incompatibleId = ProviderInstanceId.make("codex_incompatible");
	94	const compatible = yield* service.plan({
	95	projection: projection(),
	96	targetModelSelection: { instanceId: compatibleId, model: "gpt-5.1-codex" },
	97	});
	98	const incompatible = yield* service.plan({
	99	projection: projection(),
	100	targetModelSelection: { instanceId: incompatibleId, model: "gpt-5.1-codex" },
	101	});
	102	assert.equal(compatible.transition.type, "restart_and_resume");
	103	assert.equal(incompatible.transition.type, "create_with_handoff");
	104	}).pipe(
	105	Effect.provide(
	106	testLayer({
	107	[currentInstanceId]: { continuationKey: "codex:account:primary" },
	108	codex_compatible: { continuationKey: "codex:account:primary" },
	109	codex_incompatible: { continuationKey: "codex:account:other" },
	110	}),
	111	),
	112	),
	113	);

apps/server/src/orchestration-v2/ProviderSwitchService.ts+127−40

@@ -9,13 +9,21 @@ import * as Context from "effect/Context"; · 8 unmodified lines
9	import * as DateTime from "effect/DateTime";	9	import * as DateTime from "effect/DateTime";
10	import * as Effect from "effect/Effect";	10	import * as Effect from "effect/Effect";
11	import * as Layer from "effect/Layer";	11	import * as Layer from "effect/Layer";
		12	import * as Option from "effect/Option";
12	import * as Schema from "effect/Schema";	13	import * as Schema from "effect/Schema";
13		14
		15	import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
		16	import {
		17	decideProviderSessionTransition,
		18	type ProviderSessionTransition,
		19	} from "./ProviderSessionTransitionPolicy.ts";
		20
14	export interface ProviderSwitchPlanV2 {	21	export interface ProviderSwitchPlanV2 {
15	readonly instanceChanged: boolean;	22	readonly instanceChanged: boolean;
16	readonly modelChanged: boolean;	23	readonly modelChanged: boolean;
17	readonly targetProviderThreadId: ProviderThreadId \| null;	24	readonly targetProviderThreadId: ProviderThreadId \| null;
18	readonly releaseProviderSessionIds: ReadonlyArray<ProviderSessionId>;	25	readonly releaseProviderSessionIds: ReadonlyArray<ProviderSessionId>;
		26	readonly transition: ProviderSessionTransition;
19	}	27	}
20		28
21	export class ProviderSwitchPlanError extends Schema.TaggedErrorClass<ProviderSwitchPlanError>()(	29	export class ProviderSwitchPlanError extends Schema.TaggedErrorClass<ProviderSwitchPlanError>()(
@@ -39,49 +47,128 @@ export class ProviderSwitchServiceV2 extends Context.Service< · 17 unmodified lines
39	ProviderSwitchServiceV2Shape	47	ProviderSwitchServiceV2Shape
40	>()("t3/orchestration-v2/ProviderSwitchService/ProviderSwitchServiceV2") {}	48	>()("t3/orchestration-v2/ProviderSwitchService/ProviderSwitchServiceV2") {}
41		49
42	export const layer: Layer.Layer<ProviderSwitchServiceV2> = Layer.succeed(	50	export const layer: Layer.Layer<
		51	ProviderSwitchServiceV2,
		52	never,
		53	ProviderAdapterRegistry.ProviderAdapterRegistryV2
		54	> = Layer.effect(
43	ProviderSwitchServiceV2,	55	ProviderSwitchServiceV2,
44	ProviderSwitchServiceV2.of({	56	Effect.gen(function* () {
45	plan: ({ projection, targetModelSelection }) =>	57	const adapters = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
46	Effect.sync(() => {	58	return ProviderSwitchServiceV2.of({
47	const current = projection.thread.modelSelection;	59	plan: ({ projection, targetModelSelection }) =>
48	const instanceChanged = current.instanceId !== targetModelSelection.instanceId;	60	Effect.gen(function* () {
49	const modelChanged = current.model !== targetModelSelection.model;	61	const current = projection.thread.modelSelection;
50	const targetProviderThread = projection.providerThreads	62	const instanceChanged = current.instanceId !== targetModelSelection.instanceId;
51	.filter(	63	const modelChanged = current.model !== targetModelSelection.model;
52	(thread) =>	64	const getMetadata = (instanceId: typeof current.instanceId) =>
53	thread.appThreadId === projection.thread.id &&	65	adapters.getMetadata !== undefined
54	thread.ownerNodeId === null &&	66	? adapters.getMetadata(instanceId)
55	thread.providerInstanceId === targetModelSelection.instanceId,	67	: adapters.get(instanceId).pipe(
56	)	68	Effect.flatMap((adapter) =>
57	.toSorted(	69	adapter.getCapabilities().pipe(
58	(left, right) =>	70	Effect.map((capabilities) => ({
59	DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),	71	driver: adapter.driver,
60	)[0];	72	continuationKey: `${adapter.driver}:instance:${instanceId}`,
61	const releaseProviderSessionIds = projection.providerSessions	73	enabled: true,
62	.filter((session) => {	74	capabilities,
63	if (session.status === "stopped" \|\| session.status === "error") return false;	75	})),
64	if (instanceChanged) {	76	),
65	return session.providerInstanceId !== targetModelSelection.instanceId;	77	),
66	}	78	);
67	return modelChanged && !session.capabilities.sessions.supportsModelSwitchInSession;	79	const currentInstance = yield* Effect.option(getMetadata(current.instanceId));
68	})	80	const targetInstance = yield* Effect.option(getMetadata(targetModelSelection.instanceId));
69	.map((session) => session.id);	81	const currentSession = projection.providerSessions
70	return {	82	.filter((session) => session.providerInstanceId === current.instanceId)
71	instanceChanged,	83	.toSorted(
72	modelChanged,	84	(left, right) =>
73	targetProviderThreadId: targetProviderThread?.id ?? null,	85	DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
74	releaseProviderSessionIds,	86	)[0];
75	};	87	const transition = Option.isNone(targetInstance)
76	}).pipe(	88	? ({
77	Effect.mapError(	89	type: "reject",
78	(cause) =>	90	reason: "The target provider instance is unavailable.",
79	new ProviderSwitchPlanError({	91	} as const)
		92	: decideProviderSessionTransition({
		93	current:
		94	Option.isNone(currentInstance) \|\| currentSession === undefined
		95	? null
		96	: {
		97	driver: currentInstance.value.driver,
		98	continuationIdentity: {
		99	driverKind: currentInstance.value.driver,
		100	continuationKey: currentInstance.value.continuationKey,
		101	},
		102	modelSelection: current,
		103	runtimeMode: projection.thread.runtimeMode,
		104	interactionMode: projection.thread.interactionMode,
		105	workspace: currentSession.cwd,
		106	capabilities: currentSession.capabilities,
		107	},
		108	target: {
		109	driver: targetInstance.value.driver,
		110	continuationIdentity: {
		111	driverKind: targetInstance.value.driver,
		112	continuationKey: targetInstance.value.continuationKey,
		113	},
		114	modelSelection: targetModelSelection,
		115	runtimeMode: projection.thread.runtimeMode,
		116	interactionMode: projection.thread.interactionMode,
		117	workspace:
		118	projection.thread.worktreePath ??
		119	currentSession?.cwd ??
		120	"<unresolved-workspace>",
		121	capabilities: targetInstance.value.capabilities,
		122	available: targetInstance.value.enabled,
		123	},
		124	});
		125	if (transition.type === "reject") {
		126	return yield* new ProviderSwitchPlanError({
80	threadId: projection.thread.id,	127	threadId: projection.thread.id,
81	targetProviderInstanceId: targetModelSelection.instanceId,	128	targetProviderInstanceId: targetModelSelection.instanceId,
82	cause,	129	cause: transition.reason,
83	}),	130	});
		131	}
		132	const targetProviderThread = projection.providerThreads
		133	.filter(
		134	(thread) =>
		135	thread.appThreadId === projection.thread.id &&
		136	thread.ownerNodeId === null &&
		137	thread.providerInstanceId === targetModelSelection.instanceId,
		138	)
		139	.toSorted(
		140	(left, right) =>
		141	DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
		142	)[0];
		143	const releaseProviderSessionIds = projection.providerSessions
		144	.filter((session) => {
		145	if (session.status === "stopped" \|\| session.status === "error") return false;
		146	if (transition.type === "restart_and_resume") {
		147	return session.id === currentSession?.id;
		148	}
		149	if (transition.type === "create_with_handoff") {
		150	return session.providerInstanceId !== targetModelSelection.instanceId;
		151	}
		152	return false;
		153	})
		154	.map((session) => session.id);
		155	return {
		156	instanceChanged,
		157	modelChanged,
		158	targetProviderThreadId: targetProviderThread?.id ?? null,
		159	releaseProviderSessionIds,
		160	transition,
		161	};
		162	}).pipe(
		163	Effect.mapError(
		164	(cause) =>
		165	new ProviderSwitchPlanError({
		166	threadId: projection.thread.id,
		167	targetProviderInstanceId: targetModelSelection.instanceId,
		168	cause,
		169	}),
		170	),
84	),	171	),
85	),	172	});
86	}),	173	}),
87	);	174	);

apps/server/src/orchestration-v2/ProviderTurnStartService.ts+108−14

@@ -15,7 +15,10 @@ import * as Layer from "effect/Layer"; · 14 unmodified lines
15	import * as Schema from "effect/Schema";	15	import * as Schema from "effect/Schema";
16		16
17	import { EventSinkV2 } from "./EventSink.ts";	17	import { EventSinkV2 } from "./EventSink.ts";
18	import { providerMessageWithContextHandoffs } from "./ContextHandoffService.ts";	18	import {
		19	ContextHandoffServiceV2,
		20	providerMessageWithContextHandoffs,
		21	} from "./ContextHandoffService.ts";
19	import { IdAllocatorV2 } from "./IdAllocator.ts";	22	import { IdAllocatorV2 } from "./IdAllocator.ts";
20	import { ProjectionStoreV2 } from "./ProjectionStore.ts";	23	import { ProjectionStoreV2 } from "./ProjectionStore.ts";
21	import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";	24	import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
@@ -46,6 +49,7 @@ export const layer: Layer.Layer< · 24 unmodified lines
46	ProviderTurnStartServiceV2,	49	ProviderTurnStartServiceV2,
47	never,	50	never,
48	\| EventSinkV2	51	\| EventSinkV2
		52	\| ContextHandoffServiceV2
49	\| IdAllocatorV2	53	\| IdAllocatorV2
50	\| ProjectionStoreV2	54	\| ProjectionStoreV2
51	\| ProviderSessionManagerV2	55	\| ProviderSessionManagerV2
@@ -55,6 +59,7 @@ export const layer: Layer.Layer< · 3 unmodified lines
55	ProviderTurnStartServiceV2,	59	ProviderTurnStartServiceV2,
56	Effect.gen(function* () {	60	Effect.gen(function* () {
57	const eventSink = yield* EventSinkV2;	61	const eventSink = yield* EventSinkV2;
		62	const contextHandoffService = yield* ContextHandoffServiceV2;
58	const idAllocator = yield* IdAllocatorV2;	63	const idAllocator = yield* IdAllocatorV2;
59	const projectionStore = yield* ProjectionStoreV2;	64	const projectionStore = yield* ProjectionStoreV2;
60	const providerSessions = yield* ProviderSessionManagerV2;	65	const providerSessions = yield* ProviderSessionManagerV2;
@@ -95,6 +100,15 @@ export const layer: Layer.Layer< · 34 unmodified lines
95	transfer.status === "pending" &&	100	transfer.status === "pending" &&
96	transfer.resolution === null,	101	transfer.resolution === null,
97	);	102	);
		103	const existingResumeFallback = projection.contextTransfers.find(
		104	(transfer) =>
		105	transfer.type === "provider_handoff" &&
		106	transfer.sourceThreadId === projection.thread.id &&
		107	transfer.targetThreadId === projection.thread.id &&
		108	transfer.targetRunId === run.id &&
		109	transfer.status === "resolved_portable" &&
		110	transfer.resolution?.strategy === "portable_context",
		111	);
98	if (	112	if (
99	rootNode === undefined \|\|	113	rootNode === undefined \|\|
100	attempt === undefined \|\|	114	attempt === undefined \|\|
@@ -126,6 +140,7 @@ export const layer: Layer.Layer< · 25 unmodified lines
126	? {}	140	? {}
127	: { resumeFromSession: existingSessionProjection }),	141	: { resumeFromSession: existingSessionProjection }),
128	});	142	});
		143	let effectiveHandoffs = handoffs;
129	const loadedProviderThread = yield* Effect.gen(function* () {	144	const loadedProviderThread = yield* Effect.gen(function* () {
130	if (nativeForkTransfer !== undefined) {	145	if (nativeForkTransfer !== undefined) {
131	const sourceProjection = yield* projectionStore.getThreadProjection(	146	const sourceProjection = yield* projectionStore.getThreadProjection(
@@ -160,19 +175,95 @@ export const layer: Layer.Layer< · 28 unmodified lines
160	...(sourceProviderTurn === undefined ? {} : { providerTurnId: sourceProviderTurn.id }),	175	...(sourceProviderTurn === undefined ? {} : { providerTurnId: sourceProviderTurn.id }),
161	});	176	});
162	}	177	}
163	return providerThread.nativeThreadRef === null	178	if (providerThread.nativeThreadRef === null) {
164	? yield* session.ensureThread({	179	return yield* session.ensureThread({
		180	threadId: projection.thread.id,
		181	modelSelection: run.modelSelection,
		182	runtimePolicy: resolvedRuntimePolicy,
		183	providerSessionId,
		184	});
		185	}
		186	const resumed = yield* Effect.result(
		187	session.resumeThread({
		188	providerThread,
		189	threadId: projection.thread.id,
		190	modelSelection: run.modelSelection,
		191	runtimePolicy: resolvedRuntimePolicy,
		192	}),
		193	);
		194	if (resumed._tag === "Success") {
		195	return resumed.success;
		196	}
		197
		198	const replacement = yield* session.ensureThread({
		199	threadId: projection.thread.id,
		200	modelSelection: run.modelSelection,
		201	runtimePolicy: resolvedRuntimePolicy,
		202	providerSessionId,
		203	});
		204	if (existingResumeFallback !== undefined) {
		205	return replacement;
		206	}
		207	const transferId = yield* idAllocator.allocate.contextTransfer({
		208	sourceThreadId: projection.thread.id,
		209	targetThreadId: projection.thread.id,
		210	type: "provider_resume_fallback",
		211	});
		212	const createdAt = yield* DateTime.now;
		213	const handoff = yield* contextHandoffService.prepareProviderHandoff({
		214	threadId: projection.thread.id,
		215	targetRunId: run.id,
		216	transferId,
		217	fromProviderThreadIds: [providerThread.id],
		218	toProviderThreadId: providerThread.id,
		219	fromProviderInstanceId: providerThread.providerInstanceId,
		220	toProviderInstanceId: run.providerInstanceId,
		221	coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) },
		222	strategy: "full_thread_summary",
		223	items: projection.turnItems,
		224	createdAt,
		225	});
		226	effectiveHandoffs = [...handoffs, handoff];
		227	yield* eventSink.write({
		228	events: [
		229	{
		230	id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
		231	type: "context-handoff.updated",
165	threadId: projection.thread.id,	232	threadId: projection.thread.id,
166	modelSelection: run.modelSelection,	233	runId: run.id,
167	runtimePolicy: resolvedRuntimePolicy,	234	providerInstanceId: run.providerInstanceId,
168	providerSessionId,	235	occurredAt: createdAt,
169	})	236	payload: handoff,
170	: yield* session.resumeThread({	237	},
171	providerThread,	238	{
		239	id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
		240	type: "context-transfer.updated",
172	threadId: projection.thread.id,	241	threadId: projection.thread.id,
173	modelSelection: run.modelSelection,	242	runId: run.id,
174	runtimePolicy: resolvedRuntimePolicy,	243	providerInstanceId: run.providerInstanceId,
175	});	244	occurredAt: createdAt,
		245	payload: {
		246	id: transferId,
		247	type: "provider_handoff",
		248	sourceThreadId: projection.thread.id,
		249	targetThreadId: projection.thread.id,
		250	sourcePoint: { threadId: projection.thread.id },
		251	basePoint: null,
		252	sourceProviderInstanceId: providerThread.providerInstanceId,
		253	targetProviderInstanceId: run.providerInstanceId,
		254	targetRunId: run.id,
		255	status: "resolved_portable",
		256	resolution: { strategy: "portable_context", contextHandoffId: handoff.id },
		257	createdBy: "system",
		258	error: null,
		259	createdAt,
		260	updatedAt: createdAt,
		261	consumedAt: null,
		262	},
		263	},
		264	],
		265	});
		266	return replacement;
176	});	267	});
177	const now = yield* DateTime.now;	268	const now = yield* DateTime.now;
178	const runningProviderThread: OrchestrationV2ProviderThread = {	269	const runningProviderThread: OrchestrationV2ProviderThread = {
@@ -322,9 +413,12 @@ export const layer: Layer.Layer< · 143 unmodified lines
322	message: {	413	message: {
323	messageId: message.id,	414	messageId: message.id,
324	text:	415	text:
325	handoffs.length === 0	416	effectiveHandoffs.length === 0
326	? message.text	417	? message.text
327	: providerMessageWithContextHandoffs({ handoffs, userText: message.text }),	418	: providerMessageWithContextHandoffs({
		419	handoffs: effectiveHandoffs,
		420	userText: message.text,
		421	}),
328	attachments: message.attachments,	422	attachments: message.attachments,
329	createdBy: message.createdBy,	423	createdBy: message.createdBy,
330	creationSource: message.creationSource,	424	creationSource: message.creationSource,

apps/server/src/orchestration-v2/ResourceCleanupService.ts+71−0

@@ -0,0 +1,71 @@
	1	import * as Context from "effect/Context";
	2	import * as Effect from "effect/Effect";
	3	import * as FileSystem from "effect/FileSystem";
	4	import * as Layer from "effect/Layer";
	5	import * as Schema from "effect/Schema";
	6
	7	import { resolveAttachmentPathById } from "../attachmentStore.ts";
	8	import * as ServerConfig from "../config.ts";
	9	import * as TerminalManager from "../terminal/Manager.ts";
	10
	11	export class ResourceCleanupError extends Schema.TaggedErrorClass<ResourceCleanupError>()(
	12	"ResourceCleanupError",
	13	{
	14	operation: Schema.Literals(["terminal", "attachment"]),
	15	threadId: Schema.optional(Schema.String),
	16	attachmentId: Schema.optional(Schema.String),
	17	cause: Schema.Defect(),
	18	},
	19	) {}
	20
	21	export class ResourceCleanupService extends Context.Reference<{
	22	readonly cleanupTerminals: (threadId: string) => Effect.Effect<void, ResourceCleanupError>;
	23	readonly cleanupAttachments: (
	24	attachmentIds: ReadonlyArray<string>,
	25	) => Effect.Effect<void, ResourceCleanupError>;
	26	}>("t3/orchestration-v2/ResourceCleanupService", {
	27	defaultValue: () => ({
	28	cleanupTerminals: () => Effect.void,
	29	cleanupAttachments: () => Effect.void,
	30	}),
	31	}) {}
	32
	33	export const live = Layer.effect(
	34	ResourceCleanupService,
	35	Effect.gen(function* () {
	36	const terminals = yield* TerminalManager.TerminalManager;
	37	const fileSystem = yield* FileSystem.FileSystem;
	38	const config = yield* ServerConfig.ServerConfig;
	39	return {
	40	cleanupTerminals: (threadId: string) =>
	41	terminals
	42	.close({ threadId, deleteHistory: true })
	43	.pipe(
	44	Effect.mapError(
	45	(cause) => new ResourceCleanupError({ operation: "terminal", threadId, cause }),
	46	),
	47	),
	48	cleanupAttachments: (attachmentIds: ReadonlyArray<string>) =>
	49	Effect.forEach(
	50	attachmentIds,
	51	(attachmentId) => {
	52	const path = resolveAttachmentPathById({
	53	attachmentsDir: config.attachmentsDir,
	54	attachmentId,
	55	});
	56	return path === null
	57	? Effect.void
	58	: fileSystem
	59	.remove(path, { force: true })
	60	.pipe(
	61	Effect.mapError(
	62	(cause) =>
	63	new ResourceCleanupError({ operation: "attachment", attachmentId, cause }),
	64	),
	65	);
	66	},
	67	{ discard: true, concurrency: 4 },
	68	),
	69	};
	70	}),
	71	);

apps/server/src/orchestration-v2/RunFinalizationService.test.ts+41−0

@@ -0,0 +1,41 @@
	1	import { assert, it, vi } from "@effect/vitest";
	2	import {
	3	CheckpointScopeId,
	4	RunId,
	5	ThreadId,
	6	type OrchestrationV2ThreadProjection,
	7	} from "@t3tools/contracts";
	8	import * as Effect from "effect/Effect";
	9	import * as Layer from "effect/Layer";
	10
	11	import * as CheckpointCapture from "./CheckpointCaptureService.ts";
	12	import * as ProjectionStore from "./ProjectionStore.ts";
	13	import * as RunFinalization from "./RunFinalizationService.ts";
	14
	15	it.effect("captures the root checkpoint and refreshes workspace state", () => {
	16	const threadId = ThreadId.make("thread_finalize");
	17	const runId = RunId.make("run_finalize");
	18	const scopeId = CheckpointScopeId.make("scope_finalize");
	19	const capture = vi.fn(() => Effect.void);
	20	const refresh = vi.fn(() => Effect.void);
	21	const projection = {
	22	checkpointScopes: [{ id: scopeId, cwd: "/repo" }],
	23	} as unknown as OrchestrationV2ThreadProjection;
	24	const layer = RunFinalization.layer.pipe(
	25	Layer.provide(
	26	Layer.mergeAll(
	27	Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({ execute: capture }),
	28	Layer.mock(ProjectionStore.ProjectionStoreV2)({
	29	getThreadProjection: () => Effect.succeed(projection),
	30	}),
	31	Layer.succeed(RunFinalization.RunFinalizationObserver, { refresh }),
	32	),
	33	),
	34	);
	35	return Effect.gen(function* () {
	36	const service = yield* RunFinalization.RunFinalizationService;
	37	yield* service.finalize({ threadId, runId, scopeId });
	38	assert.equal(capture.mock.calls.length, 1);
	39	assert.deepEqual(refresh.mock.calls[0], ["/repo"]);
	40	}).pipe(Effect.provide(layer));
	41	});

apps/server/src/orchestration-v2/RunFinalizationService.ts+97−0

@@ -0,0 +1,97 @@
	1	import { CheckpointScopeId, RunId, ThreadId } from "@t3tools/contracts";
	2	import * as Context from "effect/Context";
	3	import * as Effect from "effect/Effect";
	4	import * as Layer from "effect/Layer";
	5	import * as Schema from "effect/Schema";
	6
	7	import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
	8	import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";
	9	import * as CheckpointCapture from "./CheckpointCaptureService.ts";
	10	import * as ProjectionStore from "./ProjectionStore.ts";
	11
	12	export class RunFinalizationError extends Schema.TaggedErrorClass<RunFinalizationError>()(
	13	"RunFinalizationError",
	14	{
	15	threadId: ThreadId,
	16	runId: RunId,
	17	scopeId: CheckpointScopeId,
	18	operation: Schema.Literals(["capture-checkpoint", "refresh-workspace"]),
	19	cause: Schema.Defect(),
	20	},
	21	) {}
	22
	23	export class RunFinalizationRefreshError extends Schema.TaggedErrorClass<RunFinalizationRefreshError>()(
	24	"RunFinalizationRefreshError",
	25	{ cwd: Schema.String, cause: Schema.Defect() },
	26	) {}
	27
	28	export class RunFinalizationObserver extends Context.Reference<{
	29	readonly refresh: (cwd: string) => Effect.Effect<void, RunFinalizationRefreshError>;
	30	}>("t3/orchestration-v2/RunFinalizationObserver", {
	31	defaultValue: () => ({ refresh: () => Effect.void }),
	32	}) {}
	33
	34	export class RunFinalizationService extends Context.Service<
	35	RunFinalizationService,
	36	{
	37	readonly finalize: (input: {
	38	readonly threadId: ThreadId;
	39	readonly runId: RunId;
	40	readonly scopeId: CheckpointScopeId;
	41	}) => Effect.Effect<void, RunFinalizationError>;
	42	}
	43	>()("t3/orchestration-v2/RunFinalizationService") {}
	44
	45	export const make = Effect.gen(function* () {
	46	const checkpointCapture = yield* CheckpointCapture.CheckpointCaptureServiceV2;
	47	const projections = yield* ProjectionStore.ProjectionStoreV2;
	48	const observer = yield* RunFinalizationObserver;
	49
	50	const finalize: RunFinalizationService["Service"]["finalize"] = Effect.fn(
	51	"RunFinalizationService.finalize",
	52	)(function* (input) {
	53	yield* checkpointCapture
	54	.execute(input)
	55	.pipe(
	56	Effect.mapError(
	57	(cause) => new RunFinalizationError({ ...input, operation: "capture-checkpoint", cause }),
	58	),
	59	);
	60	const projection = yield* projections
	61	.getThreadProjection(input.threadId)
	62	.pipe(
	63	Effect.mapError(
	64	(cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
	65	),
	66	);
	67	const cwd = projection.checkpointScopes.find((scope) => scope.id === input.scopeId)?.cwd;
	68	if (cwd !== undefined) {
	69	yield* observer
	70	.refresh(cwd)
	71	.pipe(
	72	Effect.mapError(
	73	(cause) =>
	74	new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
	75	),
	76	);
	77	}
	78	});
	79	return RunFinalizationService.of({ finalize });
	80	});
	81
	82	export const layer = Layer.effect(RunFinalizationService, make);
	83
	84	export const observerLive = Layer.effect(
	85	RunFinalizationObserver,
	86	Effect.gen(function* () {
	87	const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
	88	const vcsStatus = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
	89	return {
	90	refresh: (cwd: string) =>
	91	Effect.all([workspaceEntries.refresh(cwd), vcsStatus.refreshStatus(cwd)], {
	92	discard: true,
	93	concurrency: "unbounded",
	94	}).pipe(Effect.mapError((cause) => new RunFinalizationRefreshError({ cwd, cause }))),
	95	};
	96	}),
	97	);

apps/server/src/orchestration-v2/ThreadLaunchService.test.ts+260−0

@@ -0,0 +1,260 @@
	1	import { assert, it, vi } from "@effect/vitest";
	2	import {
	3	EventId,
	4	CommandId,
	5	ProjectId,
	6	ProviderInstanceId,
	7	ThreadId,
	8	type OrchestrationV2DomainEvent,
	9	type OrchestrationV2ThreadProjection,
	10	} from "@t3tools/contracts";
	11	import * as DateTime from "effect/DateTime";
	12	import * as Effect from "effect/Effect";
	13	import * as Layer from "effect/Layer";
	14	import * as Option from "effect/Option";
	15	import * as Stream from "effect/Stream";
	16
	17	import * as GitWorkflow from "../git/GitWorkflowService.ts";
	18	import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
	19	import * as ProjectService from "../project/ProjectService.ts";
	20	import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
	21	import * as TextGeneration from "../textGeneration/TextGeneration.ts";
	22	import * as IdAllocator from "./IdAllocator.ts";
	23	import { OrchestratorDispatchError, type OrchestratorV2DispatchResult } from "./Orchestrator.ts";
	24	import { emptyProjection } from "./ProjectionStore.ts";
	25	import * as ThreadLaunch from "./ThreadLaunchService.ts";
	26	import * as ThreadManagement from "./ThreadManagementService.ts";
	27
	28	const projectId = ProjectId.make("project_launch_test");
	29	const modelSelection = {
	30	instanceId: ProviderInstanceId.make("codex"),
	31	model: "gpt-5.1-codex",
	32	} as const;
	33	const project = {
	34	id: projectId,
	35	title: "Project",
	36	workspaceRoot: "/repo",
	37	repositoryIdentity: null,
	38	faviconPath: null,
	39	defaultModelSelection: modelSelection,
	40	scripts: [],
	41	createdAt: "2026-06-20T00:00:00.000Z",
	42	updatedAt: "2026-06-20T00:00:00.000Z",
	43	deletedAt: null,
	44	} as const;
	45
	46	function projectionFor(threadId: ThreadId): OrchestrationV2ThreadProjection {
	47	const now = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
	48	const event = {
	49	id: EventId.make(`event:${threadId}`),
	50	type: "thread.created",
	51	threadId,
	52	providerInstanceId: modelSelection.instanceId,
	53	occurredAt: now,
	54	payload: {
	55	id: threadId,
	56	projectId,
	57	title: "Thread",
	58	providerInstanceId: modelSelection.instanceId,
	59	modelSelection,
	60	runtimeMode: "full-access",
	61	interactionMode: "default",
	62	branch: null,
	63	worktreePath: null,
	64	activeProviderThreadId: null,
	65	lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
	66	forkedFrom: null,
	67	createdBy: "user",
	68	creationSource: "web",
	69	createdAt: now,
	70	updatedAt: now,
	71	archivedAt: null,
	72	deletedAt: null,
	73	},
	74	} satisfies Extract<OrchestrationV2DomainEvent, { readonly type: "thread.created" }>;
	75	return emptyProjection(event);
	76	}
	77
	78	const makeLayer = (options?: { readonly failCreate?: boolean; readonly failSetup?: boolean }) => {
	79	const projections = new Map<ThreadId, OrchestrationV2ThreadProjection>();
	80	const dispatch = vi.fn(
	81	(
	82	command: Parameters<ThreadManagement.ThreadManagementService["Service"]["dispatch"]>[0],
	83	): Effect.Effect<OrchestratorV2DispatchResult, OrchestratorDispatchError> => {
	84	if (command.type !== "thread.create") return Effect.die("unexpected command");
	85	if (options?.failCreate) {
	86	return Effect.fail(
	87	new OrchestratorDispatchError({
	88	commandId: command.commandId,
	89	commandType: command.type,
	90	cause: "create failed",
	91	}),
	92	);
	93	}
	94	projections.set(command.threadId, projectionFor(command.threadId));
	95	return Effect.succeed({ sequence: 1, storedEvents: [] });
	96	},
	97	);
	98	const createWorktree = vi.fn(() =>
	99	Effect.succeed({
	100	worktree: { path: "/repo-worktrees/feature", refName: "feature", headSha: "abc" },
	101	} as never),
	102	);
	103	const removeWorktree = vi.fn(() => Effect.void);
	104	const runForThread = vi.fn(() =>
	105	options?.failSetup
	106	? Effect.fail(
	107	new ProjectSetupScriptRunner.ProjectSetupScriptOperationError({
	108	threadId: "thread",
	109	worktreePath: "/repo-worktrees/feature",
	110	operation: "openTerminal",
	111	cause: "setup failed",
	112	}),
	113	)
	114	: Effect.succeed({ status: "no-script" as const }),
	115	);
	116	const sendToThread = vi.fn(() =>
	117	Effect.succeed({} as ThreadManagement.ThreadManagementSendResult),
	118	);
	119
	120	const dependencies = Layer.mergeAll(
	121	Layer.succeed(ProjectService.ProjectService, {
	122	create: () => Effect.die("unused"),
	123	bootstrap: () => Effect.die("unused"),
	124	update: () => Effect.die("unused"),
	125	delete: () => Effect.die("unused"),
	126	getById: (id) => Effect.succeed(id === projectId ? Option.some(project) : Option.none()),
	127	getByWorkspaceRoot: () => Effect.succeed(Option.some(project)),
	128	snapshot: Effect.die("unused"),
	129	changes: Stream.empty,
	130	}),
	131	Layer.mock(GitWorkflow.GitWorkflowService)({ createWorktree, removeWorktree }),
	132	Layer.succeed(ProjectSetupScriptRunner.ProjectSetupScriptRunner, {
	133	runForThread,
	134	}),
	135	Layer.mock(TextGeneration.TextGeneration)({
	136	generateThreadTitle: () => Effect.succeed({ title: "Generated title" }),
	137	generateBranchName: () => Effect.succeed({ branch: "generated-branch" }),
	138	}),
	139	Layer.mock(ThreadManagement.ThreadManagementService)({
	140	dispatch,
	141	getThreadProjection: (threadId) =>
	142	projections.has(threadId)
	143	? Effect.succeed(projections.get(threadId)!)
	144	: Effect.die("missing projection"),
	145	sendToThread,
	146	}),
	147	IdAllocator.layer,
	148	);
	149	return {
	150	layer: ThreadLaunch.layer.pipe(
	151	Layer.provideMerge(dependencies),
	152	Layer.provideMerge(SqlitePersistenceMemory),
	153	),
	154	dispatch,
	155	createWorktree,
	156	removeWorktree,
	157	runForThread,
	158	sendToThread,
	159	};
	160	};
	161
	162	it.effect("persists root-workspace launches and returns the committed workflow on retry", () => {
	163	const test = makeLayer();
	164	return Effect.gen(function* () {
	165	const service = yield* ThreadLaunch.ThreadLaunchService;
	166	const input = {
	167	commandId: CommandId.make("command_launch_root"),
	168	projectId,
	169	title: "Thread",
	170	modelSelection,
	171	runtimeMode: "full-access" as const,
	172	interactionMode: "default" as const,
	173	workspaceStrategy: { type: "root" as const },
	174	createdBy: "user" as const,
	175	creationSource: "web" as const,
	176	};
	177	const first = yield* service.launch(input);
	178	const retry = yield* service.launch(input);
	179	assert.equal(first.threadId, retry.threadId);
	180	assert.isFalse(first.resumed);
	181	assert.isTrue(retry.resumed);
	182	assert.equal(test.dispatch.mock.calls.length, 1);
	183	}).pipe(Effect.provide(test.layer));
	184	});
	185
	186	it.effect(
	187	"generates first-run title and branch, runs setup once, and sends the initial message",
	188	() => {
	189	const test = makeLayer();
	190	return Effect.gen(function* () {
	191	const service = yield* ThreadLaunch.ThreadLaunchService;
	192	const input = {
	193	commandId: CommandId.make("command_launch_generated"),
	194	projectId,
	195	title: "New thread",
	196	modelSelection,
	197	runtimeMode: "full-access" as const,
	198	interactionMode: "default" as const,
	199	workspaceStrategy: { type: "worktree" as const, baseRef: "main" },
	200	initialMessage: { text: "Build the feature", attachments: [] },
	201	createdBy: "user" as const,
	202	creationSource: "web" as const,
	203	};
	204	yield* service.launch(input);
	205	yield* service.launch(input);
	206	const createCommand = test.dispatch.mock.calls[0]?.[0];
	207	assert.equal(createCommand?.type, "thread.create");
	208	if (createCommand?.type === "thread.create") {
	209	assert.equal(createCommand.title, "Generated title");
	210	assert.equal(createCommand.branch, "feature");
	211	}
	212	assert.equal(test.runForThread.mock.calls.length, 1);
	213	assert.equal(test.sendToThread.mock.calls.length, 1);
	214	}).pipe(Effect.provide(test.layer));
	215	},
	216	);
	217
	218	it.effect("removes a new worktree and does not create a thread when setup fails", () => {
	219	const test = makeLayer({ failSetup: true });
	220	return Effect.gen(function* () {
	221	const service = yield* ThreadLaunch.ThreadLaunchService;
	222	yield* service
	223	.launch({
	224	commandId: CommandId.make("command_launch_setup_failure"),
	225	projectId,
	226	title: "Thread",
	227	modelSelection,
	228	runtimeMode: "full-access",
	229	interactionMode: "default",
	230	workspaceStrategy: { type: "worktree", baseRef: "main", branch: "feature" },
	231	createdBy: "user",
	232	creationSource: "web",
	233	})
	234	.pipe(Effect.flip);
	235	assert.equal(test.removeWorktree.mock.calls.length, 1);
	236	assert.equal(test.dispatch.mock.calls.length, 0);
	237	}).pipe(Effect.provide(test.layer));
	238	});
	239
	240	it.effect("compensates a newly-created worktree when thread creation fails", () => {
	241	const test = makeLayer({ failCreate: true });
	242	return Effect.gen(function* () {
	243	const service = yield* ThreadLaunch.ThreadLaunchService;
	244	yield* service
	245	.launch({
	246	commandId: CommandId.make("command_launch_worktree"),
	247	projectId,
	248	title: "Thread",
	249	modelSelection,
	250	runtimeMode: "full-access",
	251	interactionMode: "default",
	252	workspaceStrategy: { type: "worktree", baseRef: "main", branch: "feature" },
	253	createdBy: "user",
	254	creationSource: "web",
	255	})
	256	.pipe(Effect.flip);
	257	assert.equal(test.createWorktree.mock.calls.length, 1);
	258	assert.equal(test.removeWorktree.mock.calls.length, 1);
	259	}).pipe(Effect.provide(test.layer));
	260	});

apps/server/src/orchestration-v2/ThreadLaunchService.ts+410−0

@@ -0,0 +1,410 @@
	1	import {
	2	CommandId,
	3	type ChatAttachment,
	4	type ModelSelection,
	5	type OrchestrationV2Actor,
	6	type OrchestrationV2CreationSource,
	7	type OrchestrationV2ThreadProjection,
	8	type ProviderInteractionMode,
	9	ProjectId,
	10	type RuntimeMode,
	11	ThreadId,
	12	} from "@t3tools/contracts";
	13	import * as Context from "effect/Context";
	14	import * as DateTime from "effect/DateTime";
	15	import * as Effect from "effect/Effect";
	16	import * as Layer from "effect/Layer";
	17	import * as Option from "effect/Option";
	18	import * as Schema from "effect/Schema";
	19	import * as Semaphore from "effect/Semaphore";
	20	import * as SqlClient from "effect/unstable/sql/SqlClient";
	21
	22	import * as GitWorkflow from "../git/GitWorkflowService.ts";
	23	import * as ProjectService from "../project/ProjectService.ts";
	24	import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
	25	import * as TextGeneration from "../textGeneration/TextGeneration.ts";
	26	import * as IdAllocator from "./IdAllocator.ts";
	27	import * as ThreadManagement from "./ThreadManagementService.ts";
	28
	29	export type ThreadLaunchWorkspaceStrategy =
	30	\| { readonly type: "root" }
	31	\| {
	32	readonly type: "worktree";
	33	readonly baseRef: string;
	34	readonly branch?: string;
	35	};
	36
	37	export interface ThreadLaunchInitialMessage {
	38	readonly text: string;
	39	readonly attachments: ReadonlyArray<ChatAttachment>;
	40	}
	41
	42	export interface ThreadLaunchInput {
	43	readonly commandId: CommandId;
	44	readonly projectId: ProjectId;
	45	readonly title: string;
	46	readonly modelSelection: ModelSelection;
	47	readonly runtimeMode: RuntimeMode;
	48	readonly interactionMode: ProviderInteractionMode;
	49	readonly workspaceStrategy: ThreadLaunchWorkspaceStrategy;
	50	readonly initialMessage?: ThreadLaunchInitialMessage;
	51	readonly createdBy: OrchestrationV2Actor;
	52	readonly creationSource: OrchestrationV2CreationSource;
	53	}
	54
	55	export interface ThreadLaunchResult {
	56	readonly threadId: ThreadId;
	57	readonly projection: OrchestrationV2ThreadProjection;
	58	readonly resumed: boolean;
	59	}
	60
	61	export class ThreadLaunchError extends Schema.TaggedErrorClass<ThreadLaunchError>()(
	62	"ThreadLaunchError",
	63	{
	64	operation: Schema.Literals([
	65	"resolve-project",
	66	"load-workflow",
	67	"persist-workflow",
	68	"generate-metadata",
	69	"provision-worktree",
	70	"run-setup-script",
	71	"create-thread",
	72	"dispatch-message",
	73	"compensate-worktree",
	74	]),
	75	commandId: CommandId,
	76	projectId: ProjectId,
	77	threadId: Schema.optional(ThreadId),
	78	cause: Schema.Defect(),
	79	},
	80	) {
	81	override get message(): string {
	82	return `Thread launch ${this.commandId} failed during ${this.operation}.`;
	83	}
	84	}
	85
	86	type WorkflowRow = {
	87	readonly command_id: string;
	88	readonly thread_id: string;
	89	readonly project_id: string;
	90	readonly status: string;
	91	readonly title: string;
	92	readonly worktree_path: string \| null;
	93	readonly branch: string \| null;
	94	readonly setup_committed: number;
	95	readonly thread_committed: number;
	96	readonly message_committed: number;
	97	};
	98
	99	export class ThreadLaunchService extends Context.Service<
	100	ThreadLaunchService,
	101	{
	102	readonly launch: (
	103	input: ThreadLaunchInput,
	104	) => Effect.Effect<ThreadLaunchResult, ThreadLaunchError>;
	105	}
	106	>()("t3/orchestration-v2/ThreadLaunchService") {}
	107
	108	export const make = Effect.gen(function* () {
	109	const sql = yield* SqlClient.SqlClient;
	110	const projects = yield* ProjectService.ProjectService;
	111	const git = yield* GitWorkflow.GitWorkflowService;
	112	const setupScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
	113	const textGeneration = yield* TextGeneration.TextGeneration;
	114	const ids = yield* IdAllocator.IdAllocatorV2;
	115	const threads = yield* ThreadManagement.ThreadManagementService;
	116	const launchSemaphore = yield* Semaphore.make(1);
	117
	118	const mapError =
	119	(input: ThreadLaunchInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) =>
	120	(cause: unknown) =>
	121	new ThreadLaunchError({
	122	operation,
	123	commandId: input.commandId,
	124	projectId: input.projectId,
	125	...(threadId === undefined ? {} : { threadId }),
	126	cause,
	127	});
	128
	129	const loadWorkflow = Effect.fn("ThreadLaunchService.loadWorkflow")(function* (
	130	input: ThreadLaunchInput,
	131	) {
	132	const rows = yield* sql<WorkflowRow>`
	133	SELECT * FROM orchestration_v2_thread_launch_workflows
	134	WHERE command_id = ${input.commandId}
	135	`.pipe(Effect.mapError(mapError(input, "load-workflow")));
	136	return rows[0] ?? null;
	137	});
	138
	139	const saveWorkflow = Effect.fn("ThreadLaunchService.saveWorkflow")(function* (
	140	input: ThreadLaunchInput,
	141	row: {
	142	readonly threadId: ThreadId;
	143	readonly status: string;
	144	readonly title: string;
	145	readonly worktreePath: string \| null;
	146	readonly branch: string \| null;
	147	readonly setupCommitted: boolean;
	148	readonly threadCommitted: boolean;
	149	readonly messageCommitted: boolean;
	150	readonly lastError?: string \| null;
	151	},
	152	) {
	153	const now = DateTime.formatIso(yield* DateTime.now);
	154	yield* sql`
	155	INSERT INTO orchestration_v2_thread_launch_workflows (
	156	command_id, thread_id, project_id, status, title, worktree_path, branch, setup_committed,
	157	thread_committed, message_committed, last_error, created_at, updated_at
	158	) VALUES (
	159	${input.commandId}, ${row.threadId}, ${input.projectId}, ${row.status}, ${row.title},
	160	${row.worktreePath}, ${row.branch}, ${row.setupCommitted ? 1 : 0}, ${row.threadCommitted ? 1 : 0},
	161	${row.messageCommitted ? 1 : 0}, ${row.lastError ?? null}, ${now}, ${now}
	162	) ON CONFLICT(command_id) DO UPDATE SET
	163	status = excluded.status,
	164	title = excluded.title,
	165	worktree_path = excluded.worktree_path,
	166	branch = excluded.branch,
	167	setup_committed = excluded.setup_committed,
	168	thread_committed = excluded.thread_committed,
	169	message_committed = excluded.message_committed,
	170	last_error = excluded.last_error,
	171	updated_at = excluded.updated_at
	172	`.pipe(Effect.mapError(mapError(input, "persist-workflow", row.threadId)));
	173	});
	174
	175	const launchWorkflow: ThreadLaunchService["Service"]["launch"] = Effect.fn(
	176	"ThreadLaunchService.launch",
	177	)(function* (input) {
	178	const project = yield* projects.getById(input.projectId).pipe(
	179	Effect.mapError(mapError(input, "resolve-project")),
	180	Effect.flatMap(
	181	Option.match({
	182	onNone: () => Effect.fail(mapError(input, "resolve-project")("Project not found.")),
	183	onSome: Effect.succeed,
	184	}),
	185	),
	186	);
	187	const stored = yield* loadWorkflow(input);
	188	const threadId =
	189	stored === null
	190	? yield* ids.allocate
	191	.thread({ projectId: input.projectId })
	192	.pipe(Effect.mapError(mapError(input, "persist-workflow")))
	193	: ThreadId.make(stored.thread_id);
	194	if (stored?.status === "completed") {
	195	return {
	196	threadId,
	197	projection: yield* threads
	198	.getThreadProjection(threadId)
	199	.pipe(Effect.mapError(mapError(input, "create-thread", threadId))),
	200	resumed: true,
	201	};
	202	}
	203
	204	let branch = stored?.branch ?? null;
	205	let title = stored?.title ?? input.title;
	206	if (input.initialMessage !== undefined) {
	207	if (title === "New thread") {
	208	title = yield* textGeneration
	209	.generateThreadTitle({
	210	cwd: project.workspaceRoot,
	211	message: input.initialMessage.text,
	212	attachments: input.initialMessage.attachments,
	213	modelSelection: input.modelSelection,
	214	})
	215	.pipe(
	216	Effect.map((result) => result.title),
	217	Effect.mapError(mapError(input, "generate-metadata", threadId)),
	218	);
	219	}
	220	if (input.workspaceStrategy.type === "worktree" && branch === null) {
	221	branch = yield* textGeneration
	222	.generateBranchName({
	223	cwd: project.workspaceRoot,
	224	message: input.initialMessage.text,
	225	attachments: input.initialMessage.attachments,
	226	modelSelection: input.modelSelection,
	227	})
	228	.pipe(
	229	Effect.map((result) => result.branch),
	230	Effect.mapError(mapError(input, "generate-metadata", threadId)),
	231	);
	232	}
	233	}
	234	if (input.workspaceStrategy.type === "worktree") {
	235	branch = branch ?? input.workspaceStrategy.branch ?? `thread-${threadId}`;
	236	}
	237
	238	let worktreePath = stored?.worktree_path ?? null;
	239	const threadCommitted = stored?.thread_committed === 1;
	240	let setupCommitted = stored?.setup_committed === 1;
	241	let messageCommitted = stored?.message_committed === 1;
	242	if (input.workspaceStrategy.type === "worktree" && worktreePath === null) {
	243	const worktree = yield* git
	244	.createWorktree({
	245	cwd: project.workspaceRoot,
	246	refName: input.workspaceStrategy.baseRef,
	247	newRefName: branch!,
	248	baseRefName: input.workspaceStrategy.baseRef,
	249	path: null,
	250	})
	251	.pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));
	252	worktreePath = worktree.worktree.path;
	253	branch = worktree.worktree.refName;
	254	yield* saveWorkflow(input, {
	255	threadId,
	256	status: "workspace_ready",
	257	title,
	258	worktreePath,
	259	branch,
	260	setupCommitted,
	261	threadCommitted,
	262	messageCommitted,
	263	});
	264	}
	265	const cwd = worktreePath ?? project.workspaceRoot;
	266
	267	if (!setupCommitted) {
	268	const setupExit = yield* Effect.exit(
	269	setupScripts.runForThread({
	270	threadId,
	271	projectId: input.projectId,
	272	projectCwd: project.workspaceRoot,
	273	worktreePath: cwd,
	274	project: {
	275	workspaceRoot: project.workspaceRoot,
	276	scripts: project.scripts,
	277	},
	278	}),
	279	);
	280	if (setupExit._tag === "Failure") {
	281	if (worktreePath !== null && !threadCommitted) {
	282	yield* git
	283	.removeWorktree({ cwd: project.workspaceRoot, path: worktreePath })
	284	.pipe(Effect.mapError(mapError(input, "compensate-worktree", threadId)));
	285	worktreePath = null;
	286	}
	287	yield* saveWorkflow(input, {
	288	threadId,
	289	status: "setup_failed",
	290	title,
	291	worktreePath,
	292	branch,
	293	setupCommitted: false,
	294	threadCommitted,
	295	messageCommitted,
	296	lastError: "Project setup script failed.",
	297	});
	298	return yield* Effect.failCause(setupExit.cause).pipe(
	299	Effect.mapError(mapError(input, "run-setup-script", threadId)),
	300	);
	301	}
	302	setupCommitted = true;
	303	yield* saveWorkflow(input, {
	304	threadId,
	305	status: "setup_ready",
	306	title,
	307	worktreePath,
	308	branch,
	309	setupCommitted,
	310	threadCommitted,
	311	messageCommitted,
	312	});
	313	}
	314
	315	if (!threadCommitted) {
	316	const createExit = yield* Effect.exit(
	317	threads.dispatch({
	318	type: "thread.create",
	319	commandId: input.commandId,
	320	threadId,
	321	projectId: input.projectId,
	322	title,
	323	modelSelection: input.modelSelection,
	324	runtimeMode: input.runtimeMode,
	325	interactionMode: input.interactionMode,
	326	branch,
	327	worktreePath,
	328	createdBy: input.createdBy,
	329	creationSource: input.creationSource,
	330	}),
	331	);
	332	if (createExit._tag === "Failure") {
	333	if (worktreePath !== null) {
	334	yield* git
	335	.removeWorktree({ cwd: project.workspaceRoot, path: worktreePath })
	336	.pipe(Effect.mapError(mapError(input, "compensate-worktree", threadId)));
	337	worktreePath = null;
	338	setupCommitted = false;
	339	}
	340	yield* saveWorkflow(input, {
	341	threadId,
	342	status: "thread_create_failed",
	343	title,
	344	worktreePath,
	345	branch,
	346	setupCommitted,
	347	threadCommitted: false,
	348	messageCommitted,
	349	lastError: "Thread creation failed.",
	350	});
	351	return yield* Effect.failCause(createExit.cause).pipe(
	352	Effect.mapError(mapError(input, "create-thread", threadId)),
	353	);
	354	}
	355	yield* saveWorkflow(input, {
	356	threadId,
	357	status: "thread_created",
	358	title,
	359	worktreePath,
	360	branch,
	361	setupCommitted,
	362	threadCommitted: true,
	363	messageCommitted,
	364	});
	365	}
	366
	367	if (input.initialMessage !== undefined && !messageCommitted) {
	368	const messageId = yield* ids.allocate
	369	.message({ threadId, ordinal: 1 })
	370	.pipe(Effect.mapError(mapError(input, "dispatch-message", threadId)));
	371	yield* threads
	372	.sendToThread({
	373	projectId: input.projectId,
	374	commandId: CommandId.make(`${input.commandId}:initial-message`),
	375	threadId,
	376	messageId,
	377	text: input.initialMessage.text,
	378	attachments: input.initialMessage.attachments,
	379	mode: "auto",
	380	createdBy: input.createdBy,
	381	creationSource: input.creationSource,
	382	})
	383	.pipe(Effect.mapError(mapError(input, "dispatch-message", threadId)));
	384	messageCommitted = true;
	385	}
	386	yield* saveWorkflow(input, {
	387	threadId,
	388	status: "completed",
	389	title,
	390	worktreePath,
	391	branch,
	392	setupCommitted,
	393	threadCommitted: true,
	394	messageCommitted,
	395	});
	396	return {
	397	threadId,
	398	projection: yield* threads
	399	.getThreadProjection(threadId)
	400	.pipe(Effect.mapError(mapError(input, "create-thread", threadId))),
	401	resumed: stored !== null,
	402	};
	403	});
	404
	405	return ThreadLaunchService.of({
	406	launch: (input) => launchSemaphore.withPermit(launchWorkflow(input)),
	407	});
	408	});
	409
	410	export const layer = Layer.effect(ThreadLaunchService, make);

apps/server/src/orchestration-v2/ThreadLifecycleService.test.ts+60−0

@@ -0,0 +1,60 @@
	1	import { assert, it } from "@effect/vitest";
	2	import {
	3	CommandId,
	4	ProviderInstanceId,
	5	ThreadId,
	6	type OrchestrationV2ThreadProjection,
	7	} from "@t3tools/contracts";
	8	import * as Effect from "effect/Effect";
	9	import * as Layer from "effect/Layer";
	10
	11	import * as ThreadLifecycle from "./ThreadLifecycleService.ts";
	12	import * as ThreadManagement from "./ThreadManagementService.ts";
	13
	14	it.effect("maps application lifecycle operations to V2-native commands", () => {
	15	const threadId = ThreadId.make("thread_lifecycle_service");
	16	const commands: Array<string> = [];
	17	const projection = { thread: { id: threadId } } as OrchestrationV2ThreadProjection;
	18	const layer = ThreadLifecycle.layer.pipe(
	19	Layer.provide(
	20	Layer.mock(ThreadManagement.ThreadManagementService)({
	21	dispatch: (command) => {
	22	commands.push(command.type);
	23	return Effect.succeed({ sequence: commands.length, storedEvents: [] });
	24	},
	25	getThreadProjection: () => Effect.succeed(projection),
	26	}),
	27	),
	28	);
	29	return Effect.gen(function* () {
	30	const service = yield* ThreadLifecycle.ThreadLifecycleService;
	31	yield* service.archive({ commandId: CommandId.make("archive"), threadId });
	32	yield* service.unarchive({ commandId: CommandId.make("unarchive"), threadId });
	33	yield* service.updateMetadata({ commandId: CommandId.make("metadata"), threadId, title: "T" });
	34	yield* service.setRuntimeMode({
	35	commandId: CommandId.make("runtime"),
	36	threadId,
	37	runtimeMode: "approval-required",
	38	});
	39	yield* service.setInteractionMode({
	40	commandId: CommandId.make("interaction"),
	41	threadId,
	42	interactionMode: "plan",
	43	});
	44	yield* service.setModelSelection({
	45	commandId: CommandId.make("model"),
	46	threadId,
	47	modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.2" },
	48	});
	49	yield* service.delete({ commandId: CommandId.make("delete"), threadId });
	50	assert.deepEqual(commands, [
	51	"thread.archive",
	52	"thread.unarchive",
	53	"thread.metadata.update",
	54	"thread.runtime-mode.set",
	55	"thread.interaction-mode.set",
	56	"thread.model-selection.set",
	57	"thread.delete",
	58	]);
	59	}).pipe(Effect.provide(layer));
	60	});

apps/server/src/orchestration-v2/ThreadLifecycleService.ts+142−0

@@ -0,0 +1,142 @@
	1	import {
	2	CommandId,
	3	type ModelSelection,
	4	type OrchestrationV2ThreadProjection,
	5	type ProviderInteractionMode,
	6	type RuntimeMode,
	7	ThreadId,
	8	} from "@t3tools/contracts";
	9	import * as Context from "effect/Context";
	10	import * as Effect from "effect/Effect";
	11	import * as Layer from "effect/Layer";
	12	import * as Schema from "effect/Schema";
	13
	14	import * as ThreadManagement from "./ThreadManagementService.ts";
	15
	16	export class ThreadLifecycleError extends Schema.TaggedErrorClass<ThreadLifecycleError>()(
	17	"ThreadLifecycleError",
	18	{
	19	operation: Schema.Literals([
	20	"archive",
	21	"unarchive",
	22	"delete",
	23	"update-metadata",
	24	"set-runtime-mode",
	25	"set-interaction-mode",
	26	"set-model-selection",
	27	]),
	28	threadId: ThreadId,
	29	cause: Schema.Defect(),
	30	},
	31	) {
	32	override get message(): string {
	33	return `Thread lifecycle operation '${this.operation}' failed for ${this.threadId}.`;
	34	}
	35	}
	36
	37	export class ThreadLifecycleService extends Context.Service<
	38	ThreadLifecycleService,
	39	{
	40	readonly archive: (input: {
	41	readonly commandId: CommandId;
	42	readonly threadId: ThreadId;
	43	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	44	readonly unarchive: (input: {
	45	readonly commandId: CommandId;
	46	readonly threadId: ThreadId;
	47	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	48	readonly delete: (input: {
	49	readonly commandId: CommandId;
	50	readonly threadId: ThreadId;
	51	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	52	readonly updateMetadata: (input: {
	53	readonly commandId: CommandId;
	54	readonly threadId: ThreadId;
	55	readonly title?: string;
	56	readonly branch?: string \| null;
	57	readonly worktreePath?: string \| null;
	58	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	59	readonly setRuntimeMode: (input: {
	60	readonly commandId: CommandId;
	61	readonly threadId: ThreadId;
	62	readonly runtimeMode: RuntimeMode;
	63	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	64	readonly setInteractionMode: (input: {
	65	readonly commandId: CommandId;
	66	readonly threadId: ThreadId;
	67	readonly interactionMode: ProviderInteractionMode;
	68	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	69	readonly setModelSelection: (input: {
	70	readonly commandId: CommandId;
	71	readonly threadId: ThreadId;
	72	readonly modelSelection: ModelSelection;
	73	}) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadLifecycleError>;
	74	}
	75	>()("t3/orchestration-v2/ThreadLifecycleService") {}
	76
	77	export const make = Effect.gen(function* () {
	78	const threads = yield* ThreadManagement.ThreadManagementService;
	79
	80	const dispatch = <Operation extends ThreadLifecycleError["operation"]>(
	81	operation: Operation,
	82	threadId: ThreadId,
	83	command: Parameters<ThreadManagement.ThreadManagementService["Service"]["dispatch"]>[0],
	84	) =>
	85	threads.dispatch(command).pipe(
	86	Effect.andThen(threads.getThreadProjection(threadId)),
	87	Effect.mapError((cause) => new ThreadLifecycleError({ operation, threadId, cause })),
	88	);
	89
	90	return ThreadLifecycleService.of({
	91	archive: (input) =>
	92	dispatch("archive", input.threadId, {
	93	type: "thread.archive",
	94	commandId: input.commandId,
	95	threadId: input.threadId,
	96	}),
	97	unarchive: (input) =>
	98	dispatch("unarchive", input.threadId, {
	99	type: "thread.unarchive",
	100	commandId: input.commandId,
	101	threadId: input.threadId,
	102	}),
	103	delete: (input) =>
	104	dispatch("delete", input.threadId, {
	105	type: "thread.delete",
	106	commandId: input.commandId,
	107	threadId: input.threadId,
	108	}),
	109	updateMetadata: (input) =>
	110	dispatch("update-metadata", input.threadId, {
	111	type: "thread.metadata.update",
	112	commandId: input.commandId,
	113	threadId: input.threadId,
	114	...(input.title === undefined ? {} : { title: input.title }),
	115	...(input.branch === undefined ? {} : { branch: input.branch }),
	116	...(input.worktreePath === undefined ? {} : { worktreePath: input.worktreePath }),
	117	}),
	118	setRuntimeMode: (input) =>
	119	dispatch("set-runtime-mode", input.threadId, {
	120	type: "thread.runtime-mode.set",
	121	commandId: input.commandId,
	122	threadId: input.threadId,
	123	runtimeMode: input.runtimeMode,
	124	}),
	125	setInteractionMode: (input) =>
	126	dispatch("set-interaction-mode", input.threadId, {
	127	type: "thread.interaction-mode.set",
	128	commandId: input.commandId,
	129	threadId: input.threadId,
	130	interactionMode: input.interactionMode,
	131	}),
	132	setModelSelection: (input) =>
	133	dispatch("set-model-selection", input.threadId, {
	134	type: "thread.model-selection.set",
	135	commandId: input.commandId,
	136	threadId: input.threadId,
	137	modelSelection: input.modelSelection,
	138	}),
	139	});
	140	});
	141
	142	export const layer = Layer.effect(ThreadLifecycleService, make);

apps/server/src/orchestration-v2/applicationLayer.ts+23−0

@@ -0,0 +1,23 @@
	1	import * as Layer from "effect/Layer";
	2
	3	import { ProjectionProjectRepositoryLive } from "../persistence/Layers/ProjectionProjects.ts";
	4	import { layer as projectServiceLayer } from "../project/ProjectService.ts";
	5	import { layer as threadLaunchServiceLayer } from "./ThreadLaunchService.ts";
	6	import { layer as threadLifecycleServiceLayer } from "./ThreadLifecycleService.ts";
	7	import { live as resourceCleanupLive } from "./ResourceCleanupService.ts";
	8	import { observerLive as runFinalizationObserverLive } from "./RunFinalizationService.ts";
	9
	10	const projectServiceProvided = projectServiceLayer.pipe(
	11	Layer.provide(ProjectionProjectRepositoryLive),
	12	);
	13
	14	const applicationServices = Layer.mergeAll(
	15	threadLaunchServiceLayer,
	16	threadLifecycleServiceLayer,
	17	).pipe(Layer.provideMerge(projectServiceProvided));
	18
	19	export const OrchestrationV2ApplicationLayer = Layer.mergeAll(
	20	applicationServices,
	21	resourceCleanupLive,
	22	runFinalizationObserverLive,
	23	);

apps/server/src/orchestration-v2/runtimeLayer.test.ts+2−1

 import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
 import { OrchestrationV2LayerLive } from "./runtimeLayer.ts";
 import { shellStreamItemFromSnapshot } from "./ShellStream.ts";
+import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
 const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
   prefix: "t3-orchestration-v2-runtime-layer-",
 const orchestrationAdapter = {
   instanceId: modelSelection.instanceId,
   driver,
-  getCapabilities: () => Effect.die("capabilities are not used by lifecycle tests"),
+  getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
   openSession: () => Effect.die("sessions are not used by lifecycle tests"),
 } as ProviderAdapterV2Shape;
 const providerInstance = {

apps/server/src/orchestration-v2/runtimeLayer.ts+33−3

 import * as Layer from "effect/Layer";
+import * as Effect from "effect/Effect";
 import { layer as checkpointCaptureServiceLayer } from "./CheckpointCaptureService.ts";
 import { layer as checkpointServiceLayer } from "./CheckpointService.ts";
 import { layer as checkpointRollbackServiceLayer } from "./CheckpointRollbackService.ts";
 import { layerFromProviderInstanceRegistry as providerAdapterRegistryLayerFromProviderInstances } from "./ProviderAdapterRegistry.ts";
 import { layer as providerEventIngestorLayer } from "./ProviderEventIngestor.ts";
 import { layer as providerSessionManagerLayer } from "./ProviderSessionManager.ts";
+import {
+  layer as providerRuntimeRecoveryLayer,
+  ProviderRuntimeRecoveryService,
+} from "./ProviderRuntimeRecoveryService.ts";
 import { layer as providerSwitchServiceLayer } from "./ProviderSwitchService.ts";
 import { layer as providerTurnControlServiceLayer } from "./ProviderTurnControlService.ts";
 import { layer as providerTurnStartServiceLayer } from "./ProviderTurnStartService.ts";
 import { layer as runExecutionServiceLayer } from "./RunExecutionService.ts";
+import { layer as runFinalizationServiceLayer } from "./RunFinalizationService.ts";
 import { layer as runtimePolicyLayer } from "./RuntimePolicy.ts";
 import { layer as runtimeRequestServiceLayer } from "./RuntimeRequestService.ts";
 import { layer as threadManagementServiceLayer } from "./ThreadManagementService.ts";
 );
 const providerAdapterRegistryProvided = providerAdapterRegistryLayerFromProviderInstances;
+const providerSwitchServiceProvided = providerSwitchServiceLayer.pipe(
+  Layer.provide(providerAdapterRegistryProvided),
+);
 const providerAdapterRegistryProvided = providerAdapterRegistryLayerFromEffect(
   Effect.all([
 const providerTurnStartServiceProvided = providerTurnStartServiceLayer.pipe(
   Layer.provide(
     Layer.mergeAll(
+      contextHandoffServiceProvided,
       eventSinkProvided,
       idAllocatorLayer,
       projectionStoreLayer,
     ),
   ),
 );
+const runFinalizationServiceProvided = runFinalizationServiceLayer.pipe(
+  Layer.provide(Layer.merge(checkpointCaptureServiceProvided, projectionStoreLayer)),
+);
 const effectExecutorProvided = effectExecutorLayer.pipe(
   Layer.provide(
     Layer.mergeAll(
-      checkpointCaptureServiceProvided,
+      runFinalizationServiceProvided,
       checkpointRollbackServiceProvided,
       providerSessionManagerProvided,
       providerTurnControlServiceProvided,
   Layer.provide(effectWorkerProvided),
 );
+const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe(
+  Layer.provide(
+    Layer.mergeAll(
+      effectWorkerProvided,
+      storesLayer,
+      eventSinkProvided,
+      idAllocatorLayer,
+      projectionStoreLayer,
+      providerSessionManagerProvided,
+    ),
+  ),
+);
+const providerRuntimeRecoveryStartupProvided = Layer.effectDiscard(
+  Effect.flatMap(ProviderRuntimeRecoveryService, (recovery) => recovery.recover),
+).pipe(Layer.provide(providerRuntimeRecoveryProvided));
 const orchestratorProvided = orchestratorLayer.pipe(
   Layer.provide(
     Layer.mergeAll(
       providerEventIngestorProvided,
       runtimePolicyLayer,
       providerSessionManagerProvided,
-      providerSwitchServiceLayer,
+      providerSwitchServiceProvided,
       runExecutionServiceProvided,
       threadForkServiceLayer,
     ),
   ),
 );
 export const OrchestrationV2LayerLive = Layer.mergeAll(
   orchestratorProvided,
   threadManagementServiceLayer.pipe(Layer.provide(orchestratorProvided)),
   effectWorkerDaemonProvided,
+  providerRuntimeRecoveryStartupProvided,
+  providerRuntimeRecoveryProvided,
   projectionMaintenanceProvided,
 );

apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts+10−2

 import { layer as providerTurnControlServiceLayer } from "../ProviderTurnControlService.ts";
 import { layer as providerTurnStartServiceLayer } from "../ProviderTurnStartService.ts";
 import { layer as runExecutionServiceLayer } from "../RunExecutionService.ts";
+import { layer as runFinalizationServiceLayer } from "../RunFinalizationService.ts";
 import {
   layer as runtimePolicyLayer,
   layerWithOverride as runtimePolicyLayerWithOverride,
   const providerSessionManagerProvided = providerSessionManagerLayer.pipe(
     Layer.provide(Layer.mergeAll(registryLayer, eventSinkProvided, idAllocatorLayer, storesLayer)),
   );
+  const providerSwitchServiceProvided = providerSwitchServiceLayer.pipe(
+    Layer.provide(registryLayer),
+  );
   const runExecutionServiceProvided = runExecutionServiceLayer.pipe(
     Layer.provide(
       Layer.mergeAll(
   const providerTurnStartServiceProvided = providerTurnStartServiceLayer.pipe(
     Layer.provide(
       Layer.mergeAll(
+        contextHandoffServiceProvided,
         eventSinkProvided,
         idAllocatorLayer,
         storesLayer,
       Layer.mergeAll(checkpointServiceProvided, eventSinkProvided, idAllocatorLayer, storesLayer),
     ),
   );
+  const runFinalizationServiceProvided = runFinalizationServiceLayer.pipe(
+    Layer.provide(Layer.merge(checkpointCaptureServiceProvided, storesLayer)),
+  );
   const effectExecutorProvided = effectExecutorLayer.pipe(
     Layer.provide(
       Layer.mergeAll(
-        checkpointCaptureServiceProvided,
+        runFinalizationServiceProvided,
         checkpointRollbackServiceProvided,
         providerSessionManagerProvided,
         providerTurnControlServiceProvided,
         registryLayer,
         runtimeLayer,
         providerSessionManagerProvided,
-        providerSwitchServiceLayer,
+        providerSwitchServiceProvided,
         runExecutionServiceProvided,
         threadForkServiceLayer,
       ),

apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts+7−2

   readonly responseByRunOrdinal: Readonly<Record<number, string>>;
   readonly responseByThreadId?: Readonly<Record<string, Readonly<Record<number, string>>>>;
   readonly capturedTurns: Ref.Ref<ReadonlyArray<CapturedTurn>>;
+  readonly failResume?: boolean;
 }): ProviderAdapterV2Shape {
   return {
     instanceId: input.instanceId,
                 updatedAt: createdAt,
               } satisfies OrchestrationV2ProviderThread;
             }),
-          resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
+          resumeThread: ({ providerThread }) =>
+            input.failResume
+              ? unimplemented(input.driver, "simulated native resume failure")
+              : Effect.succeed(providerThread),
           startTurn: (turnInput) =>
             Effect.gen(function* () {
               yield* Effect.yieldNow;
 });
 describe("orchestration v2 provider switching", () => {
-  it.live("switches Codex to Claude and returns to the original Codex thread with handoffs", () =>
+  it.live("uses portable fallback when native resume fails after a provider switch", () =>
     Effect.scoped(
       Effect.gen(function* () {
         const cwd = yield* checkpointWorkspace("provider-switch");
               3: "codex after return",
             },
             capturedTurns,
+            failResume: true,
           }),
           makeTestAdapter({
             instanceId: ProviderInstanceId.make("claudeAgent"),

apps/server/src/persistence/Migrations.ts+2−0

 import Migration0034 from "./Migrations/034_OrchestrationV2Subagents.ts";
 import Migration0035 from "./Migrations/035_OrchestrationV2Foundation.ts";
 import Migration0036 from "./Migrations/036_OrchestrationV2ProviderSessionBindings.ts";
+import Migration0037 from "./Migrations/037_OrchestrationV2ThreadLaunchWorkflows.ts";
 /**
  * Migration loader with all migrations defined inline.
   [34, "OrchestrationV2Subagents", Migration0034],
   [35, "OrchestrationV2Foundation", Migration0035],
   [36, "OrchestrationV2ProviderSessionBindings", Migration0036],
+  [37, "OrchestrationV2ThreadLaunchWorkflows", Migration0037],
 ] as const;
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/037_OrchestrationV2ThreadLaunchWorkflows.ts+23−0

@@ -0,0 +1,23 @@
	1	import * as Effect from "effect/Effect";
	2	import * as SqlClient from "effect/unstable/sql/SqlClient";
	3
	4	export default Effect.gen(function* () {
	5	const sql = yield* SqlClient.SqlClient;
	6	yield* sql`
	7	CREATE TABLE IF NOT EXISTS orchestration_v2_thread_launch_workflows (
	8	command_id TEXT PRIMARY KEY,
	9	thread_id TEXT NOT NULL,
	10	project_id TEXT NOT NULL,
	11	status TEXT NOT NULL,
	12	title TEXT NOT NULL,
	13	worktree_path TEXT,
	14	branch TEXT,
	15	setup_committed INTEGER NOT NULL DEFAULT 0,
	16	thread_committed INTEGER NOT NULL DEFAULT 0,
	17	message_committed INTEGER NOT NULL DEFAULT 0,
	18	last_error TEXT,
	19	created_at TEXT NOT NULL,
	20	updated_at TEXT NOT NULL
	21	)
	22	`;
	23	});

apps/server/src/project/ProjectService.test.ts+20−0

@@ -129,4 +129,24 @@ it.layer(TestLayer)("ProjectService", (it) => { · 128 unmodified lines
129	assert.equal(error._tag, "ProjectConflictError");	129	assert.equal(error._tag, "ProjectConflictError");
130	}),	130	}),
131	);	131	);
		132
		133	it.effect("auto-bootstraps a workspace exactly once", () =>
		134	Effect.gen(function* () {
		135	const service = yield* ProjectService.ProjectService;
		136	yield* TestClock.setTime(Date.parse("2026-06-20T10:00:00.000Z"));
		137	const input = {
		138	projectId: ProjectId.make("project:bootstrap"),
		139	title: "Bootstrap",
		140	workspaceRoot: "/work/bootstrap/",
		141	};
		142	const first = yield* service.bootstrap(input);
		143	const second = yield* service.bootstrap({
		144	...input,
		145	projectId: ProjectId.make("project:bootstrap:unused"),
		146	});
		147	assert.isTrue(first.created);
		148	assert.isFalse(second.created);
		149	assert.equal(second.project.id, first.project.id);
		150	}),
		151	);
132	});	152	});

apps/server/src/project/ProjectService.ts+17−0

   readonly scripts?: ReadonlyArray<ProjectScript>;
 }
+export interface ProjectBootstrapInput extends ProjectCreateInput {}
 export class ProjectNotFoundError extends Schema.TaggedErrorClass<ProjectNotFoundError>()(
   "ProjectNotFoundError",
   { projectId: ProjectId },
   ProjectService,
   {
     readonly create: (input: ProjectCreateInput) => Effect.Effect<Project, ProjectServiceError>;
+    readonly bootstrap: (
+      input: ProjectBootstrapInput,
+    ) => Effect.Effect<
+      { readonly project: Project; readonly created: boolean },
+      ProjectServiceError
+    >;
     readonly update: (input: ProjectUpdateInput) => Effect.Effect<Project, ProjectServiceError>;
     readonly delete: (projectId: ProjectId) => Effect.Effect<Project, ProjectServiceError>;
     readonly getById: (
     },
   );
+  const bootstrap: ProjectService["Service"]["bootstrap"] = Effect.fn("ProjectService.bootstrap")(
+    function* (input) {
+      const existing = yield* getByWorkspaceRoot(input.workspaceRoot);
+      if (Option.isSome(existing)) return { project: existing.value, created: false };
+      return { project: yield* create(input), created: true };
+    },
+  );
   const deleteProject: ProjectService["Service"]["delete"] = Effect.fn("ProjectService.delete")(
     function* (projectId) {
       const existing = yield* projects
   return ProjectService.of({
     create,
+    bootstrap,
     update,
     delete: deleteProject,
     getById,

apps/server/src/project/ProjectSetupScriptRunner.ts+22−14

-import { ProjectId } from "@t3tools/contracts";
+import { ProjectId, type ProjectScript } from "@t3tools/contracts";
 import { projectScriptRuntimeEnv, setupProjectScript } from "@t3tools/shared/projectScripts";
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
   readonly projectCwd?: string;
   readonly worktreePath: string;
   readonly preferredTerminalId?: string;
+  readonly project?: {
+    readonly workspaceRoot: string;
+    readonly scripts: ReadonlyArray<ProjectScript>;
+  };
 }
 export class ProjectSetupScriptOperationError extends Schema.TaggedErrorClass<ProjectSetupScriptOperationError>()(
       ...(input.projectId === undefined ? {} : { projectId: input.projectId }),
       ...(input.projectCwd === undefined ? {} : { projectCwd: input.projectCwd }),
     };
-    const projectById = input.projectId
+    const suppliedProject = input.project;
-      ? yield* projectionSnapshotQuery.getProjectShellById(ProjectId.make(input.projectId)).pipe(
+    const projectById =
-          Effect.map(Option.getOrUndefined),
+      suppliedProject ??
-          Effect.mapError(
+      (input.projectId
-            (cause) =>
+        ? yield* projectionSnapshotQuery.getProjectShellById(ProjectId.make(input.projectId)).pipe(
-              new ProjectSetupScriptOperationError({
+            Effect.map(Option.getOrUndefined),
-                ...errorContext,
+            Effect.mapError(
-                operation: "resolveProject",
+              (cause) =>
-                cause,
+                new ProjectSetupScriptOperationError({
-              }),
+                  ...errorContext,
-          ),
+                  operation: "resolveProject",
-        )
+                  cause,
-      : null;
+                }),
+            ),
+          )
+        : null);
     const project =
+      suppliedProject ??
       projectById ??
       (input.projectCwd
         ? yield* projectionSnapshotQuery.getActiveProjectByWorkspaceRoot(input.projectCwd).pipe(