History

fix(orchestrator): hydrate shell cache and group multi-environment projects (#3640)

Restore versioned shell-cache hydration, reconcile repository enrichment across multiple environments, and decode non-null settled shell timestamps correctly.

Mike Olson committed 3 months ago

Commit 1bc99ec1 parent 480ec8417 files · +1759 −51Browse files

Showing the changes to packages/client-runtime/src (6 of 17 files). Show all

Unified · Split

6 files
  1. packages/client-runtime/src/platform/orchestrationCache.test.ts +87 −13
  2. packages/client-runtime/src/platform/orchestrationCache.ts +8 −4
  3. packages/client-runtime/src/state/shell-sync.test.ts +418 −3
  4. packages/client-runtime/src/state/shell.ts +48 −11
  5. packages/client-runtime/src/state/shellReducer.test.ts +325 −1
  6. packages/client-runtime/src/state/shellReducer.ts +89 −2
packages/client-runtime/src/platform/orchestrationCache.test.ts+87−13
@@ -1,5 +1,11 @@
1import { describe, expect, it } from "@effect/vitest";1import { describe, expect, it } from "@effect/vitest";
2import { EnvironmentId } from "@t3tools/contracts";2import {
3 EnvironmentId,
4 MessageId,
5 RuntimeRequestId,
6 type OrchestrationV2ShellSnapshot,
7} from "@t3tools/contracts";
8import * as DateTime from "effect/DateTime";
3import * as Effect from "effect/Effect";9import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";10import * as Option from "effect/Option";
5import * as Schema from "effect/Schema";11import * as Schema from "effect/Schema";
@@ -10,39 +16,107 @@ import { · 4 unmodified lines
10 StoredOrchestrationThreadSnapshot,16 StoredOrchestrationThreadSnapshot,
11 decodeOrDiscardOrchestrationCache,17 decodeOrDiscardOrchestrationCache,
12} from "./orchestrationCache.ts";18} from "./orchestrationCache.ts";
13import { v2Projection, v2ShellSnapshot, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts";19import {
20 v2Now,
21 v2Projection,
22 v2ShellSnapshot,
23 v2ThreadShell,
24 v2ThreadId,
25} from "../state/orchestrationV2TestFixtures.ts";
14 26
15const environmentId = EnvironmentId.make("environment-cache-test");27const environmentId = EnvironmentId.make("environment-cache-test");
16const decodeStoredShellSnapshot = Schema.decodeUnknownEffect(StoredOrchestrationShellSnapshot);28const StoredShellSnapshotJson = Schema.fromJsonString(StoredOrchestrationShellSnapshot);
17const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect(StoredOrchestrationThreadSnapshot);29const StoredThreadSnapshotJson = Schema.fromJsonString(StoredOrchestrationThreadSnapshot);
30const encodeStoredShellSnapshot = Schema.encodeSync(StoredOrchestrationShellSnapshot);
31const decodeStoredShellSnapshotSync = Schema.decodeUnknownSync(StoredOrchestrationShellSnapshot);
32const encodeStoredShellSnapshotJson = Schema.encodeSync(StoredShellSnapshotJson);
33const decodeStoredShellSnapshotJson = Schema.decodeUnknownSync(StoredShellSnapshotJson);
34const encodeStoredThreadSnapshotJson = Schema.encodeSync(StoredThreadSnapshotJson);
35const decodeStoredThreadSnapshotJson = Schema.decodeUnknownSync(StoredThreadSnapshotJson);
36
37class TestCacheDecodeError extends Schema.TaggedErrorClass<TestCacheDecodeError>()(
38 "TestCacheDecodeError",
39 {
40 message: Schema.String,
41 },
42) {}
43
44const shellSnapshotWithSummaries: OrchestrationV2ShellSnapshot = {
45 ...v2ShellSnapshot,
46 threads: [
47 {
48 ...v2ThreadShell,
49 pendingRuntimeRequest: {
50 id: RuntimeRequestId.make("runtime-request-v2"),
51 kind: "command",
52 createdAt: v2Now,
53 },
54 latestVisibleMessage: {
55 id: MessageId.make("message-v2"),
56 role: "assistant",
57 text: "Done",
58 updatedAt: v2Now,
59 },
60 latestUserMessageAt: v2Now,
61 settledAt: v2Now,
62 },
63 ],
64};
18 65
19describe("orchestration cache envelopes", () => {66describe("orchestration cache envelopes", () => {
20 it.effect("accepts V2 shell and thread cache envelopes", () =>67 it.effect("round-trips V2 shell and thread cache envelopes as JSON", () =>
21 Effect.gen(function* () {68 Effect.sync(() => {
22 const shell = yield* decodeStoredShellSnapshot({69 const encodedShell = encodeStoredShellSnapshotJson({
23 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,70 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
24 environmentId,71 environmentId,
25 snapshot: v2ShellSnapshot,72 snapshot: shellSnapshotWithSummaries,
26 });73 });
27 const thread = yield* decodeStoredThreadSnapshot({74 const shell = decodeStoredShellSnapshotJson(encodedShell);
75 const encodedThread = encodeStoredThreadSnapshotJson({
28 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,76 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
29 environmentId,77 environmentId,
30 threadId: v2ThreadId,78 threadId: v2ThreadId,
31 thread: v2Projection,79 thread: v2Projection,
32 });80 });
81 const thread = decodeStoredThreadSnapshotJson(encodedThread);
82 const [threadShell] = shell.snapshot.threads;
33 83
34 expect(shell.snapshot).toEqual(v2ShellSnapshot);84 expect(threadShell?.latestVisibleMessage?.text).toBe("Done");
35 expect(thread.thread).toEqual(v2Projection);85 expect(
86 threadShell?.latestVisibleMessage === null ||
87 threadShell?.latestVisibleMessage === undefined
88 ? null
89 : DateTime.formatIso(threadShell.latestVisibleMessage.updatedAt),
90 ).toBe("2026-06-20T00:00:00.000Z");
91 expect(
92 threadShell?.latestUserMessageAt === null || threadShell?.latestUserMessageAt === undefined
93 ? null
94 : DateTime.formatIso(threadShell.latestUserMessageAt),
95 ).toBe("2026-06-20T00:00:00.000Z");
96 expect(
97 threadShell?.settledAt === null || threadShell?.settledAt === undefined
98 ? null
99 : DateTime.formatIso(threadShell.settledAt),
100 ).toBe("2026-06-20T00:00:00.000Z");
101 expect(DateTime.formatIso(thread.snapshot.projection.thread.updatedAt)).toBe(
102 "2026-06-20T00:00:00.000Z",
103 );
104 expect(thread.snapshot.projection).toEqual(v2Projection);
105 expect(thread.snapshot.snapshotSequence).toBe(4);
36 }),106 }),
37 );107 );
38 108
39 it.effect("discards V1-versioned cache envelopes after a decode failure", () =>109 it.effect("discards V1-versioned cache envelopes after a decode failure", () =>
40 Effect.gen(function* () {110 Effect.gen(function* () {
41 let discardCount = 0;111 let discardCount = 0;
42 const decoded = decodeStoredShellSnapshot({112 const encodedShell = encodeStoredShellSnapshot({
43 schemaVersion: 1,113 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
44 environmentId,114 environmentId,
45 snapshot: v2ShellSnapshot,115 snapshot: v2ShellSnapshot,
116 });
117 const decoded = Effect.try({
118 try: () => decodeStoredShellSnapshotSync({ ...encodedShell, schemaVersion: 1 }),
119 catch: (cause) => new TestCacheDecodeError({ message: String(cause) }),
46 }).pipe(Effect.map(Option.some));120 }).pipe(Effect.map(Option.some));
47 121
48 const result = yield* decodeOrDiscardOrchestrationCache(122 const result = yield* decodeOrDiscardOrchestrationCache(
packages/client-runtime/src/platform/orchestrationCache.ts+8−4
@@ -1,7 +1,8 @@
1import {1import {
2 EnvironmentId,2 EnvironmentId,
3 OrchestrationV2ShellSnapshot,3 OrchestrationV2ShellSnapshotJson,
4 OrchestrationV2ThreadProjection,4 OrchestrationV2ThreadDetailSnapshot,
5 OrchestrationV2ThreadProjectionJson,
5 ThreadId,6 ThreadId,
6} from "@t3tools/contracts";7} from "@t3tools/contracts";
7import * as Effect from "effect/Effect";8import * as Effect from "effect/Effect";
@@ -13,14 +14,17 @@ export const ORCHESTRATION_CACHE_SCHEMA_VERSION = 2 as const; · 5 unmodified lines
13export const StoredOrchestrationShellSnapshot = Schema.Struct({14export const StoredOrchestrationShellSnapshot = Schema.Struct({
14 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),15 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),
15 environmentId: EnvironmentId,16 environmentId: EnvironmentId,
16 snapshot: OrchestrationV2ShellSnapshot,17 snapshot: OrchestrationV2ShellSnapshotJson,
17});18});
18 19
19export const StoredOrchestrationThreadSnapshot = Schema.Struct({20export const StoredOrchestrationThreadSnapshot = Schema.Struct({
20 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),21 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),
21 environmentId: EnvironmentId,22 environmentId: EnvironmentId,
22 threadId: ThreadId,23 threadId: ThreadId,
23 thread: OrchestrationV2ThreadProjection,24 snapshot: OrchestrationV2ThreadDetailSnapshot.mapFields((fields) => ({
25 ...fields,
26 projection: OrchestrationV2ThreadProjectionJson,
27 })),
24});28});
25 29
26/** Invalid orchestration caches are disposable and must never block live synchronization. */30/** Invalid orchestration caches are disposable and must never block live synchronization. */
packages/client-runtime/src/state/shell-sync.test.ts+418−3
@@ -5,10 +5,14 @@ import { · 4 unmodified lines
5 type OrchestrationV2ShellStreamItem,5 type OrchestrationV2ShellStreamItem,
6} from "@t3tools/contracts";6} from "@t3tools/contracts";
7import { describe, expect, it } from "@effect/vitest";7import { describe, expect, it } from "@effect/vitest";
8import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";9import * as Effect from "effect/Effect";
10import * as Exit from "effect/Exit";
11import * as Fiber from "effect/Fiber";
9import * as Option from "effect/Option";12import * as Option from "effect/Option";
10import * as Queue from "effect/Queue";13import * as Queue from "effect/Queue";
11import * as Ref from "effect/Ref";14import * as Ref from "effect/Ref";
15import * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";16import * as Stream from "effect/Stream";
13import * as SubscriptionRef from "effect/SubscriptionRef";17import * as SubscriptionRef from "effect/SubscriptionRef";
14 18
@@ -22,8 +26,8 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; · 7 unmodified lines
22import * as Persistence from "../platform/persistence.ts";26import * as Persistence from "../platform/persistence.ts";
23import * as RpcSession from "../rpc/session.ts";27import * as RpcSession from "../rpc/session.ts";
24import type { WsRpcProtocolClient } from "../rpc/protocol.ts";28import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
25import { makeEnvironmentShellState } from "./shell.ts";29import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts";
26import { v2ShellSnapshot } from "./orchestrationV2TestFixtures.ts";30import { v2Project, v2ShellSnapshot } from "./orchestrationV2TestFixtures.ts";
27 31
28const TARGET = new PrimaryConnectionTarget({32const TARGET = new PrimaryConnectionTarget({
29 environmentId: EnvironmentId.make("environment-1"),33 environmentId: EnvironmentId.make("environment-1"),
@@ -70,7 +74,7 @@ describe("environment shell synchronization", () => { · 40 unmodified lines
70 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);74 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
71 const cache = Persistence.EnvironmentCacheStore.of({75 const cache = Persistence.EnvironmentCacheStore.of({
72 loadShell: () => Effect.succeed(Option.none()),76 loadShell: () => Effect.succeed(Option.none()),
73 saveShell: () => Effect.never,77 saveShell: () => Effect.void,
74 loadThread: () => Effect.succeed(Option.none()),78 loadThread: () => Effect.succeed(Option.none()),
75 saveThread: () => Effect.void,79 saveThread: () => Effect.void,
76 removeThread: () => Effect.void,80 removeThread: () => Effect.void,
@@ -352,4 +356,415 @@ describe("environment shell synchronization", () => { · 275 unmodified lines
352 expect(yield* Ref.get(loaderCalls)).toBe(2);356 expect(yield* Ref.get(loaderCalls)).toBe(2);
353 }),357 }),
354 );358 );
359
360 it.effect("flushes the latest live snapshot without blocking connection state updates", () =>
361 Effect.gen(function* () {
362 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
363 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
364 const saveStarted = yield* Deferred.make<void>();
365 const releaseSave = yield* Deferred.make<void>();
366 const client = {
367 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
368 } as unknown as WsRpcProtocolClient;
369 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
370 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
371 Option.some(session(client)),
372 );
373 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
374 target: TARGET,
375 state: supervisorState,
376 session: activeSession,
377 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
378 connect: Effect.void,
379 disconnect: Effect.void,
380 retryNow: Effect.void,
381 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
382 const cache = Persistence.EnvironmentCacheStore.of({
383 loadShell: () => Effect.succeed(Option.none()),
384 saveShell: (_environmentId, snapshot) =>
385 Deferred.succeed(saveStarted, undefined).pipe(
386 Effect.andThen(Deferred.await(releaseSave)),
387 Effect.andThen(Ref.update(saved, (values) => [...values, snapshot])),
388 ),
389 loadThread: () => Effect.succeed(Option.none()),
390 saveThread: () => Effect.void,
391 removeThread: () => Effect.void,
392 loadServerConfig: () => Effect.succeed(Option.none()),
393 saveServerConfig: () => Effect.void,
394 loadVcsRefs: () => Effect.succeed(Option.none()),
395 saveVcsRefs: () => Effect.void,
396 clear: () => Effect.void,
397 });
398 const snapshotLoader = ShellSnapshotLoader.of({
399 load: () => Effect.succeed(Option.none()),
400 });
401
402 const shellState = yield* makeEnvironmentShellState().pipe(
403 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
404 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
405 Effect.provideService(ShellSnapshotLoader, snapshotLoader),
406 );
407
408 yield* SubscriptionRef.set(supervisorState, {
409 desired: true,
410 network: "online",
411 phase: "connecting",
412 stage: "synchronizing",
413 attempt: 1,
414 generation: 0,
415 lastFailure: null,
416 retryAt: null,
417 });
418 yield* Queue.offer(events, {
419 kind: "snapshot",
420 snapshot: LIVE_SHELL_SNAPSHOT,
421 });
422 yield* Queue.offer(events, { kind: "synchronized" });
423 yield* SubscriptionRef.changes(shellState).pipe(
424 Stream.filter((state) => state.status === "live"),
425 Stream.runHead,
426 );
427
428 yield* SubscriptionRef.set(supervisorState, {
429 desired: true,
430 network: "online",
431 phase: "connecting",
432 stage: "synchronizing",
433 attempt: 2,
434 generation: 1,
435 lastFailure: null,
436 retryAt: null,
437 });
438 yield* SubscriptionRef.set(supervisorState, {
439 desired: false,
440 network: "online",
441 phase: "available",
442 stage: null,
443 attempt: 2,
444 generation: 2,
445 lastFailure: null,
446 retryAt: null,
447 });
448 yield* Deferred.await(saveStarted);
449
450 yield* SubscriptionRef.set(supervisorState, {
451 desired: true,
452 network: "online",
453 phase: "connecting",
454 stage: "synchronizing",
455 attempt: 3,
456 generation: 3,
457 lastFailure: null,
458 retryAt: null,
459 });
460 for (let index = 0; index < 100; index += 1) {
461 if ((yield* SubscriptionRef.get(shellState)).status === "synchronizing") break;
462 yield* Effect.yieldNow;
463 }
464
465 expect((yield* SubscriptionRef.get(shellState)).status).toBe("synchronizing");
466 expect(yield* Ref.get(saved)).toEqual([]);
467
468 yield* Deferred.succeed(releaseSave, undefined);
469 for (let index = 0; index < 100; index += 1) {
470 if ((yield* Ref.get(saved)).length > 0) break;
471 yield* Effect.yieldNow;
472 }
473
474 expect(yield* Ref.get(saved)).toEqual([LIVE_SHELL_SNAPSHOT]);
475 }),
476 );
477
478 it.effect("re-persists same-sequence enrichment when an older in-flight save completes", () =>
479 Effect.gen(function* () {
480 const repositoryIdentity = {
481 canonicalKey: "github.com/example/repo",
482 locator: {
483 source: "git-remote" as const,
484 remoteName: "origin",
485 remoteUrl: "https://github.com/example/repo.git",
486 },
487 };
488 const unenrichedSnapshot: OrchestrationV2ShellSnapshot = {
489 ...LIVE_SHELL_SNAPSHOT,
490 snapshotSequence: 1,
491 projects: [{ ...v2Project, repositoryIdentity: null }],
492 };
493 const enrichedSnapshot: OrchestrationV2ShellSnapshot = {
494 ...LIVE_SHELL_SNAPSHOT,
495 snapshotSequence: 1,
496 projects: [{ ...v2Project, repositoryIdentity }],
497 };
498
499 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
500 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
501 const unenrichedSaveStarted = yield* Deferred.make<void>();
502 const releaseUnenrichedSave = yield* Deferred.make<void>();
503 const client = {
504 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
505 } as unknown as WsRpcProtocolClient;
506 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
507 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
508 Option.some(session(client)),
509 );
510 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
511 target: TARGET,
512 state: supervisorState,
513 session: activeSession,
514 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
515 connect: Effect.void,
516 disconnect: Effect.void,
517 retryNow: Effect.void,
518 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
519 const cache = Persistence.EnvironmentCacheStore.of({
520 loadShell: () => Effect.succeed(Option.none()),
521 saveShell: (_environmentId, snapshot) =>
522 Effect.gen(function* () {
523 if (snapshot.projects[0]?.repositoryIdentity == null) {
524 yield* Deferred.succeed(unenrichedSaveStarted, undefined);
525 yield* Deferred.await(releaseUnenrichedSave);
526 }
527 yield* Ref.update(saved, (values) => [...values, snapshot]);
528 }),
529 loadThread: () => Effect.succeed(Option.none()),
530 saveThread: () => Effect.void,
531 removeThread: () => Effect.void,
532 loadServerConfig: () => Effect.succeed(Option.none()),
533 saveServerConfig: () => Effect.void,
534 loadVcsRefs: () => Effect.succeed(Option.none()),
535 saveVcsRefs: () => Effect.void,
536 clear: () => Effect.void,
537 });
538 const snapshotLoader = ShellSnapshotLoader.of({
539 load: () => Effect.succeed(Option.none()),
540 });
541
542 const shellState = yield* makeEnvironmentShellState().pipe(
543 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
544 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
545 Effect.provideService(ShellSnapshotLoader, snapshotLoader),
546 );
547
548 yield* SubscriptionRef.set(supervisorState, {
549 desired: true,
550 network: "online",
551 phase: "connecting",
552 stage: "synchronizing",
553 attempt: 1,
554 generation: 0,
555 lastFailure: null,
556 retryAt: null,
557 });
558 yield* Queue.offer(events, {
559 kind: "snapshot",
560 snapshot: unenrichedSnapshot,
561 });
562 yield* Queue.offer(events, { kind: "synchronized" });
563 yield* SubscriptionRef.changes(shellState).pipe(
564 Stream.filter((state) => state.status === "live"),
565 Stream.runHead,
566 );
567
568 // Disconnect flush starts an immediate persist of the unenriched snapshot
569 // (debounced path needs TestClock advancement; flush does not).
570 yield* SubscriptionRef.set(supervisorState, {
571 desired: false,
572 network: "online",
573 phase: "available",
574 stage: null,
575 attempt: 1,
576 generation: 1,
577 lastFailure: null,
578 retryAt: null,
579 });
580 yield* Deferred.await(unenrichedSaveStarted);
581
582 // Same sequence, different content. Session stays open so the stream can
583 // still apply enrichment while the older save is in flight.
584 yield* Queue.offer(events, {
585 kind: "snapshot",
586 snapshot: enrichedSnapshot,
587 });
588 yield* SubscriptionRef.changes(shellState).pipe(
589 Stream.filter(
590 (state) =>
591 Option.isSome(state.snapshot) &&
592 state.snapshot.value.projects[0]?.repositoryIdentity != null,
593 ),
594 Stream.runHead,
595 );
596
597 // Completing the older same-sequence save must re-persist the latest object.
598 yield* Deferred.succeed(releaseUnenrichedSave, undefined);
599 for (let index = 0; index < 100; index += 1) {
600 const values = yield* Ref.get(saved);
601 if (values.length > 0 && values.at(-1)?.projects[0]?.repositoryIdentity != null) {
602 break;
603 }
604 yield* Effect.yieldNow;
605 }
606
607 const finalSaved = (yield* Ref.get(saved)).at(-1);
608 expect(finalSaved?.snapshotSequence).toBe(1);
609 expect(finalSaved?.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
610 }),
611 );
612
613 it.effect("scope teardown flush stays stable while a save is blocked", () =>
614 Effect.gen(function* () {
615 const lateSnapshot: OrchestrationV2ShellSnapshot = {
616 ...LIVE_SHELL_SNAPSHOT,
617 snapshotSequence: 2,
618 projects: [{ ...v2Project, title: "Late stream event" }],
619 };
620 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
621 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
622 const saveStarted = yield* Deferred.make<void>();
623 const releaseSave = yield* Deferred.make<void>();
624 const client = {
625 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
626 } as unknown as WsRpcProtocolClient;
627 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
628 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
629 Option.some(session(client)),
630 );
631 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
632 target: TARGET,
633 state: supervisorState,
634 session: activeSession,
635 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
636 connect: Effect.void,
637 disconnect: Effect.void,
638 retryNow: Effect.void,
639 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
640 const cache = Persistence.EnvironmentCacheStore.of({
641 loadShell: () => Effect.succeed(Option.none()),
642 saveShell: (_environmentId, snapshot) =>
643 Deferred.succeed(saveStarted, undefined).pipe(
644 Effect.andThen(Deferred.await(releaseSave)),
645 Effect.andThen(Ref.update(saved, (values) => [...values, snapshot])),
646 ),
647 loadThread: () => Effect.succeed(Option.none()),
648 saveThread: () => Effect.void,
649 removeThread: () => Effect.void,
650 loadServerConfig: () => Effect.succeed(Option.none()),
651 saveServerConfig: () => Effect.void,
652 loadVcsRefs: () => Effect.succeed(Option.none()),
653 saveVcsRefs: () => Effect.void,
654 clear: () => Effect.void,
655 });
656 const snapshotLoader = ShellSnapshotLoader.of({
657 load: () => Effect.succeed(Option.none()),
658 });
659
660 const scope = yield* Scope.make();
661 const shellState = yield* makeEnvironmentShellState().pipe(
662 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
663 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
664 Effect.provideService(ShellSnapshotLoader, snapshotLoader),
665 Scope.provide(scope),
666 );
667
668 yield* SubscriptionRef.set(supervisorState, {
669 desired: true,
670 network: "online",
671 phase: "connecting",
672 stage: "synchronizing",
673 attempt: 1,
674 generation: 0,
675 lastFailure: null,
676 retryAt: null,
677 });
678 yield* Queue.offer(events, {
679 kind: "snapshot",
680 snapshot: LIVE_SHELL_SNAPSHOT,
681 });
682 yield* Queue.offer(events, { kind: "synchronized" });
683 yield* SubscriptionRef.changes(shellState).pipe(
684 Stream.filter((state) => state.status === "live"),
685 Stream.runHead,
686 );
687
688 // Close scope so the finalizer flush blocks on save. Workers must be
689 // interrupted before that flush; late stream events must not extend it.
690 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void));
691 yield* Deferred.await(saveStarted);
692 yield* Queue.offer(events, {
693 kind: "snapshot",
694 snapshot: lateSnapshot,
695 });
696 for (let index = 0; index < 50; index += 1) {
697 yield* Effect.yieldNow;
698 }
699 yield* Deferred.succeed(releaseSave, undefined);
700 yield* Fiber.join(closeFiber);
701
702 const values = yield* Ref.get(saved);
703 expect(values.length).toBeGreaterThan(0);
704 expect(values.every((snapshot) => snapshot.snapshotSequence === 1)).toBe(true);
705 expect(values.every((snapshot) => snapshot.projects[0]?.title !== "Late stream event")).toBe(
706 true,
707 );
708 }),
709 );
710
711 it.effect("applies authoritative lower-sequence HTTP resets over client-ahead cache", () =>
712 Effect.gen(function* () {
713 const cachedSnapshot: OrchestrationV2ShellSnapshot = {
714 ...v2ShellSnapshot,
715 snapshotSequence: 20,
716 projects: [{ ...v2Project, title: "Client ahead" }],
717 };
718 const httpSnapshot: OrchestrationV2ShellSnapshot = {
719 ...v2ShellSnapshot,
720 snapshotSequence: 4,
721 projects: [{ ...v2Project, title: "Server reset" }],
722 };
723 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
724 const client = {
725 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
726 } as unknown as WsRpcProtocolClient;
727 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
728 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
729 target: TARGET,
730 state: supervisorState,
731 session: yield* SubscriptionRef.make(Option.some(session(client))),
732 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
733 connect: Effect.void,
734 disconnect: Effect.void,
735 retryNow: Effect.void,
736 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
737 const cache = Persistence.EnvironmentCacheStore.of({
738 loadShell: () => Effect.succeed(Option.some(cachedSnapshot)),
739 saveShell: () => Effect.void,
740 loadThread: () => Effect.succeed(Option.none()),
741 saveThread: () => Effect.void,
742 removeThread: () => Effect.void,
743 loadServerConfig: () => Effect.succeed(Option.none()),
744 saveServerConfig: () => Effect.void,
745 loadVcsRefs: () => Effect.succeed(Option.none()),
746 saveVcsRefs: () => Effect.void,
747 clear: () => Effect.void,
748 });
749 const snapshotLoader = ShellSnapshotLoader.of({
750 load: () => Effect.succeed(Option.some(httpSnapshot)),
751 });
752 const shellState = yield* makeEnvironmentShellState().pipe(
753 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
754 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
755 Effect.provideService(ShellSnapshotLoader, snapshotLoader),
756 );
757
758 yield* SubscriptionRef.changes(shellState).pipe(
759 Stream.filter(
760 (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 4,
761 ),
762 Stream.runHead,
763 );
764
765 const state = yield* SubscriptionRef.get(shellState);
766 expect(Option.getOrThrow(state.snapshot).snapshotSequence).toBe(4);
767 expect(Option.getOrThrow(state.snapshot).projects[0]?.title).toBe("Server reset");
768 }),
769 );
355});770});
packages/client-runtime/src/state/shell.ts+48−11
@@ -24,7 +24,7 @@ import { EnvironmentCacheStore } from "../platform/persistence.ts"; · 23 unmodified lines
24import { subscribeDynamic } from "../rpc/client.ts";24import { subscribeDynamic } from "../rpc/client.ts";
25import type { RpcSession } from "../rpc/session.ts";25import type { RpcSession } from "../rpc/session.ts";
26import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";26import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";
27import { applyShellStreamEvent } from "./shellReducer.ts";27import { applyShellStreamEvent, mergeShellSnapshotProjects } from "./shellReducer.ts";
28import type { EnvironmentCatalogState } from "./connections.ts";28import type { EnvironmentCatalogState } from "./connections.ts";
29import { followStreamInEnvironment } from "./runtime.ts";29import { followStreamInEnvironment } from "./runtime.ts";
30 30
@@ -72,23 +72,49 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") · 41 unmodified lines
72 status: shellStatusForSnapshot(cachedSnapshot),72 status: shellStatusForSnapshot(cachedSnapshot),
73 error: Option.none(),73 error: Option.none(),
74 });74 });
75 const awaitingCompletion = yield* Ref.make(false);
76 const latestLiveSnapshot = yield* Ref.make<Option.Option<OrchestrationV2ShellSnapshot>>(
77 Option.none(),
78 );
75 const persistence = yield* Queue.sliding<OrchestrationV2ShellSnapshot>(1);79 const persistence = yield* Queue.sliding<OrchestrationV2ShellSnapshot>(1);
76 80
77 const persist = Effect.fn("EnvironmentShellState.persist")(function* (81 const persist = Effect.fn("EnvironmentShellState.persist")(function* (
78 snapshot: OrchestrationV2ShellSnapshot,82 snapshot: OrchestrationV2ShellSnapshot,
79 ) {83 ) {
80 yield* cache.saveShell(environmentId, snapshot).pipe(84 let nextSnapshot = snapshot;
81 Effect.catch((error) =>85 while (true) {
82 Effect.logWarning("Could not persist environment shell cache.").pipe(86 yield* cache.saveShell(environmentId, nextSnapshot).pipe(
83 Effect.annotateLogs({87 Effect.catch((error) =>
84 environmentId,88 Effect.logWarning("Could not persist environment shell cache.").pipe(
85 ...safeErrorLogAttributes(error),89 Effect.annotateLogs({
86 }),90 environmentId,
91 ...safeErrorLogAttributes(error),
92 }),
93 ),
87 ),94 ),
88 ),95 );
89 );96
97 const latestSnapshot = yield* Ref.get(latestLiveSnapshot);
98 // Same sequence can still mean newer content (e.g. repository enrichment).
99 if (Option.isNone(latestSnapshot) || latestSnapshot.value === nextSnapshot) {
100 return;
101 }
102 nextSnapshot = latestSnapshot.value;
103 }
104 });
105
106 const flushLiveShellSnapshot = Effect.gen(function* () {
107 const snapshot = yield* Ref.get(latestLiveSnapshot);
108 if (Option.isNone(snapshot)) {
109 return;
110 }
111 yield* persist(snapshot.value);
90 });112 });
91 113
114 // Register before scoped worker fibers so reverse finalizer order interrupts
115 // those fibers first and this flush sees a stable latestLiveSnapshot.
116 yield* Effect.addFinalizer(() => flushLiveShellSnapshot);
117
92 yield* Stream.fromQueue(persistence).pipe(118 yield* Stream.fromQueue(persistence).pipe(
93 Stream.debounce("500 millis"),119 Stream.debounce("500 millis"),
94 Stream.runForEach(persist),120 Stream.runForEach(persist),
@@ -102,6 +128,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") · 7 unmodified lines
102 status: shellStatusForSnapshot(current.snapshot),128 status: shellStatusForSnapshot(current.snapshot),
103 })),129 })),
104 ),130 ),
131 Effect.andThen(flushLiveShellSnapshot.pipe(Effect.forkScoped)),
132 Effect.asVoid,
105 );133 );
106 const setSynchronizing = SubscriptionRef.update(state, (current) => ({134 const setSynchronizing = SubscriptionRef.update(state, (current) => ({
107 ...current,135 ...current,
@@ -149,7 +177,15 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") · 41 unmodified lines
149 const current = yield* SubscriptionRef.get(state);177 const current = yield* SubscriptionRef.get(state);
150 const nextSnapshot =178 const nextSnapshot =
151 item.kind === "snapshot"179 item.kind === "snapshot"
152 ? item.snapshot180 ? mergeShellSnapshotProjects(
181 Option.getOrNull(current.snapshot),
182 item.snapshot,
183 item.resolvedRepositoryIdentityRoots === undefined
184 ? undefined
185 : {
186 resolvedRepositoryIdentityRoots: item.resolvedRepositoryIdentityRoots,
187 },
188 )
153 : Option.match(current.snapshot, {189 : Option.match(current.snapshot, {
154 onNone: () => null,190 onNone: () => null,
155 onSome: (snapshot) =>191 onSome: (snapshot) =>
@@ -161,6 +197,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") · 5 unmodified lines
161 return;197 return;
162 }198 }
163 199
200 yield* Ref.set(latestLiveSnapshot, Option.some(nextSnapshot));
164 const waiting = yield* Ref.get(awaitingCompletion);201 const waiting = yield* Ref.get(awaitingCompletion);
165 yield* SubscriptionRef.set(state, {202 yield* SubscriptionRef.set(state, {
166 snapshot: Option.some(nextSnapshot),203 snapshot: Option.some(nextSnapshot),
packages/client-runtime/src/state/shellReducer.test.ts+325−1
@@ -2,7 +2,25 @@ import { ProjectId, ThreadId } from "@t3tools/contracts"; · 1 unmodified line
2import { describe, expect, it } from "vite-plus/test";2import { describe, expect, it } from "vite-plus/test";
3 3
4import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";4import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
5import { applyShellStreamEvent } from "./shellReducer.ts";5import { applyShellStreamEvent, mergeShellSnapshotProjects } from "./shellReducer.ts";
6
7const repositoryIdentity = {
8 canonicalKey: "github.com/example/repo",
9 locator: {
10 source: "git-remote" as const,
11 remoteName: "origin",
12 remoteUrl: "https://github.com/example/repo.git",
13 },
14};
15
16const otherIdentity = {
17 canonicalKey: "github.com/example/other",
18 locator: {
19 source: "git-remote" as const,
20 remoteName: "origin",
21 remoteUrl: "https://github.com/example/other.git",
22 },
23};
6 24
7describe("applyShellStreamEvent", () => {25describe("applyShellStreamEvent", () => {
8 it("ignores stale project updates without mutating the snapshot", () => {26 it("ignores stale project updates without mutating the snapshot", () => {
@@ -42,6 +60,312 @@ describe("applyShellStreamEvent", () => { · 33 unmodified lines
42 expect(removed.projects).toEqual([]);60 expect(removed.projects).toEqual([]);
43 });61 });
44 62
63 it("keeps prior repositoryIdentity when a delta arrives with null identity", () => {
64 const withIdentity = applyShellStreamEvent(v2ShellSnapshot, {
65 kind: "project.updated",
66 sequence: 1,
67 project: { ...v2Project, repositoryIdentity },
68 });
69 expect(withIdentity.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
70
71 const withoutIdentity = applyShellStreamEvent(withIdentity, {
72 kind: "project.updated",
73 sequence: 2,
74 project: { ...v2Project, title: "Still same root", repositoryIdentity: null },
75 });
76 expect(withoutIdentity.projects[0]?.title).toBe("Still same root");
77 expect(withoutIdentity.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
78 });
79
80 it("does not keep prior repositoryIdentity after a workspace root move", () => {
81 const withIdentity = applyShellStreamEvent(v2ShellSnapshot, {
82 kind: "project.updated",
83 sequence: 1,
84 project: { ...v2Project, repositoryIdentity },
85 });
86
87 const moved = applyShellStreamEvent(withIdentity, {
88 kind: "project.updated",
89 sequence: 2,
90 project: {
91 ...v2Project,
92 workspaceRoot: "/tmp/other-root",
93 repositoryIdentity: null,
94 },
95 });
96 expect(moved.projects[0]?.workspaceRoot).toBe("/tmp/other-root");
97 expect(moved.projects[0]?.repositoryIdentity).toBeNull();
98 });
99
100 it("merges full snapshot projects without dropping known repositoryIdentity", () => {
101 const previous = {
102 ...v2ShellSnapshot,
103 projects: [{ ...v2Project, repositoryIdentity }],
104 };
105 const next = mergeShellSnapshotProjects(previous, {
106 ...v2ShellSnapshot,
107 snapshotSequence: 2,
108 projects: [{ ...v2Project, title: "Refreshed", repositoryIdentity: null }],
109 });
110 expect(next.projects[0]?.title).toBe("Refreshed");
111 expect(next.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
112 });
113
114 it("authoritative lower-sequence reset replaces client-ahead structural state", () => {
115 const previous = {
116 ...v2ShellSnapshot,
117 snapshotSequence: 9,
118 projects: [{ ...v2Project, title: "Client ahead", repositoryIdentity }],
119 threads: [{ ...v2ThreadShell, title: "Local-only thread" }],
120 };
121 const next = mergeShellSnapshotProjects(previous, {
122 ...v2ShellSnapshot,
123 snapshotSequence: 3,
124 projects: [{ ...v2Project, title: "Server reset", repositoryIdentity: null }],
125 threads: [{ ...v2ThreadShell, title: "Server thread" }],
126 });
127
128 expect(next.snapshotSequence).toBe(3);
129 expect(next.projects[0]?.title).toBe("Server reset");
130 expect(next.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
131 expect(next.threads[0]?.title).toBe("Server thread");
132 });
133
134 it("client-ahead survives lower marked enrichment then still resets on unmarked authoritative", () => {
135 const previous = {
136 ...v2ShellSnapshot,
137 snapshotSequence: 20,
138 projects: [{ ...v2Project, title: "Client ahead", repositoryIdentity }],
139 threads: [{ ...v2ThreadShell, title: "Local-only thread" }],
140 };
141 const serverSnapshot = {
142 ...v2ShellSnapshot,
143 snapshotSequence: 3,
144 projects: [{ ...v2Project, title: "Server reset", repositoryIdentity: null }],
145 threads: [{ ...v2ThreadShell, title: "Server thread" }],
146 };
147
148 // Lower-sequence marked enrichment is identity-only and must not replace structure.
149 const afterEnrichment = mergeShellSnapshotProjects(previous, serverSnapshot, {
150 resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot],
151 });
152 expect(afterEnrichment.snapshotSequence).toBe(20);
153 expect(afterEnrichment.projects[0]?.title).toBe("Client ahead");
154 expect(afterEnrichment.projects[0]?.repositoryIdentity).toBeNull();
155 expect(afterEnrichment.threads[0]?.title).toBe("Local-only thread");
156
157 // Unmarked authoritative at the same lower sequence still performs the reset.
158 const afterAuthoritative = mergeShellSnapshotProjects(afterEnrichment, serverSnapshot);
159 expect(afterAuthoritative.snapshotSequence).toBe(3);
160 expect(afterAuthoritative.projects[0]?.title).toBe("Server reset");
161 expect(afterAuthoritative.projects[0]?.repositoryIdentity).toBeNull();
162 expect(afterAuthoritative.threads[0]?.title).toBe("Server thread");
163 });
164
165 it("authoritative null retains identity then same-sequence marked enrichment clears it", () => {
166 const previous = {
167 ...v2ShellSnapshot,
168 snapshotSequence: 9,
169 projects: [{ ...v2Project, title: "Client ahead", repositoryIdentity }],
170 threads: [{ ...v2ThreadShell, title: "Local-only thread" }],
171 };
172 const serverSnapshot = {
173 ...v2ShellSnapshot,
174 snapshotSequence: 3,
175 projects: [{ ...v2Project, title: "Server reset", repositoryIdentity: null }],
176 threads: [{ ...v2ThreadShell, title: "Server thread" }],
177 };
178
179 const afterAuthoritative = mergeShellSnapshotProjects(previous, serverSnapshot);
180 expect(afterAuthoritative.snapshotSequence).toBe(3);
181 expect(afterAuthoritative.projects[0]?.title).toBe("Server reset");
182 expect(afterAuthoritative.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
183 expect(afterAuthoritative.threads[0]?.title).toBe("Server thread");
184
185 const afterEnrichment = mergeShellSnapshotProjects(afterAuthoritative, serverSnapshot, {
186 resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot],
187 });
188 expect(afterEnrichment.snapshotSequence).toBe(3);
189 expect(afterEnrichment.projects[0]?.title).toBe("Server reset");
190 expect(afterEnrichment.projects[0]?.repositoryIdentity).toBeNull();
191 expect(afterEnrichment.threads[0]?.title).toBe("Server thread");
192 });
193
194 it("takes identity from a lower-sequence enrichment without reverting live thread state", () => {
195 const previous = {
196 ...v2ShellSnapshot,
197 snapshotSequence: 5,
198 projects: [{ ...v2Project, repositoryIdentity: null }],
199 threads: [{ ...v2ThreadShell, title: "Newest title" }],
200 };
201 const next = mergeShellSnapshotProjects(
202 previous,
203 {
204 ...v2ShellSnapshot,
205 snapshotSequence: 4,
206 projects: [{ ...v2Project, repositoryIdentity }],
207 threads: [{ ...v2ThreadShell, title: "Stale title" }],
208 },
209 { resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot] },
210 );
211
212 expect(next.snapshotSequence).toBe(5);
213 expect(next.threads[0]?.title).toBe("Newest title");
214 expect(next.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
215 });
216
217 it("lower-sequence enrichment cannot resurrect empty structural state", () => {
218 const previous = {
219 ...v2ShellSnapshot,
220 snapshotSequence: 7,
221 projects: [],
222 threads: [],
223 archivedThreads: [],
224 };
225 const next = mergeShellSnapshotProjects(
226 previous,
227 {
228 ...v2ShellSnapshot,
229 snapshotSequence: 4,
230 projects: [{ ...v2Project, repositoryIdentity }],
231 threads: [v2ThreadShell],
232 },
233 { resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot] },
234 );
235
236 expect(next.snapshotSequence).toBe(7);
237 expect(next.projects).toEqual([]);
238 expect(next.threads).toEqual([]);
239 });
240
241 it("resolved null clears identity only for the marked root", () => {
242 const otherProject = {
243 ...v2Project,
244 id: ProjectId.make("project-other"),
245 workspaceRoot: "/workspace/other",
246 repositoryIdentity: otherIdentity,
247 };
248 const previous = {
249 ...v2ShellSnapshot,
250 snapshotSequence: 5,
251 projects: [{ ...v2Project, repositoryIdentity }, otherProject],
252 };
253 const next = mergeShellSnapshotProjects(
254 previous,
255 {
256 ...v2ShellSnapshot,
257 snapshotSequence: 4,
258 projects: [
259 { ...v2Project, repositoryIdentity: null },
260 { ...otherProject, repositoryIdentity: null },
261 ],
262 },
263 { resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot] },
264 );
265
266 expect(next.snapshotSequence).toBe(5);
267 expect(next.projects).toHaveLength(2);
268 expect(next.projects[0]?.repositoryIdentity).toBeNull();
269 expect(next.projects[1]?.repositoryIdentity).toEqual(otherIdentity);
270 });
271
272 it("unrelated unresolved roots retain identity in the same enrichment snapshot", () => {
273 const otherProject = {
274 ...v2Project,
275 id: ProjectId.make("project-other"),
276 workspaceRoot: "/workspace/other",
277 repositoryIdentity: otherIdentity,
278 };
279 const previous = {
280 ...v2ShellSnapshot,
281 snapshotSequence: 3,
282 projects: [{ ...v2Project, repositoryIdentity }, otherProject],
283 threads: [{ ...v2ThreadShell, title: "Current thread" }],
284 };
285 const next = mergeShellSnapshotProjects(
286 previous,
287 {
288 ...v2ShellSnapshot,
289 snapshotSequence: 3,
290 projects: [
291 { ...v2Project, repositoryIdentity },
292 { ...otherProject, repositoryIdentity: null },
293 ],
294 threads: [{ ...v2ThreadShell, title: "Stale thread" }],
295 },
296 { resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot] },
297 );
298
299 expect(next.threads[0]?.title).toBe("Current thread");
300 expect(next.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
301 expect(next.projects[1]?.repositoryIdentity).toEqual(otherIdentity);
302 });
303
304 it("clears identity for multiple roots resolved in one enrichment batch", () => {
305 const otherProject = {
306 ...v2Project,
307 id: ProjectId.make("project-other"),
308 workspaceRoot: "/workspace/other",
309 repositoryIdentity: otherIdentity,
310 };
311 const previous = {
312 ...v2ShellSnapshot,
313 snapshotSequence: 3,
314 projects: [{ ...v2Project, repositoryIdentity }, otherProject],
315 };
316 const next = mergeShellSnapshotProjects(
317 previous,
318 {
319 ...v2ShellSnapshot,
320 snapshotSequence: 3,
321 projects: [
322 { ...v2Project, repositoryIdentity: null },
323 { ...otherProject, repositoryIdentity: null },
324 ],
325 },
326 {
327 resolvedRepositoryIdentityRoots: [v2Project.workspaceRoot, otherProject.workspaceRoot],
328 },
329 );
330
331 expect(next.projects.map((project) => project.repositoryIdentity)).toEqual([null, null]);
332 });
333
334 it("does not replace a live identity with one from an unmarked lower-sequence enrichment", () => {
335 const liveIdentity = {
336 canonicalKey: "github.com/example/live",
337 locator: {
338 source: "git-remote" as const,
339 remoteName: "origin",
340 remoteUrl: "https://github.com/example/live.git",
341 },
342 };
343 const staleIdentity = {
344 canonicalKey: "github.com/example/stale",
345 locator: {
346 source: "git-remote" as const,
347 remoteName: "origin",
348 remoteUrl: "https://github.com/example/stale.git",
349 },
350 };
351 const next = mergeShellSnapshotProjects(
352 {
353 ...v2ShellSnapshot,
354 snapshotSequence: 5,
355 projects: [{ ...v2Project, repositoryIdentity: liveIdentity }],
356 },
357 {
358 ...v2ShellSnapshot,
359 snapshotSequence: 4,
360 projects: [{ ...v2Project, repositoryIdentity: staleIdentity }],
361 },
362 { resolvedRepositoryIdentityRoots: [] },
363 );
364
365 expect(next.snapshotSequence).toBe(5);
366 expect(next.projects[0]?.repositoryIdentity).toEqual(liveIdentity);
367 });
368
45 it("moves a thread between active and archive without duplicating it", () => {369 it("moves a thread between active and archive without duplicating it", () => {
46 const archived = applyShellStreamEvent(v2ShellSnapshot, {370 const archived = applyShellStreamEvent(v2ShellSnapshot, {
47 kind: "thread.updated",371 kind: "thread.updated",
packages/client-runtime/src/state/shellReducer.ts+89−2
@@ -1,4 +1,5 @@
1import type {1import type {
2 OrchestrationProjectShell,
2 OrchestrationV2ShellSnapshot,3 OrchestrationV2ShellSnapshot,
3 OrchestrationV2ShellStreamItem,4 OrchestrationV2ShellStreamItem,
4} from "@t3tools/contracts";5} from "@t3tools/contracts";
@@ -12,6 +13,85 @@ function upsertById<T extends { readonly id: unknown }>( · 7 unmodified lines
12 return items.map((candidate, candidateIndex) => (candidateIndex === index ? item : candidate));13 return items.map((candidate, candidateIndex) => (candidateIndex === index ? item : candidate));
13}14}
14 15
16function retainRepositoryIdentity(
17 previous: OrchestrationProjectShell | undefined,
18 next: OrchestrationProjectShell,
19): OrchestrationProjectShell {
20 if (
21 next.repositoryIdentity == null &&
22 previous?.repositoryIdentity != null &&
23 previous.workspaceRoot === next.workspaceRoot
24 ) {
25 return { ...next, repositoryIdentity: previous.repositoryIdentity };
26 }
27 return next;
28}
29
30export interface MergeShellSnapshotOptions {
31 /**
32 * Enrichment refresh: structure and sequence must not roll back when at or
33 * below cache; listed roots accept identity exactly (including null).
34 * Omit this options object for authoritative HTTP/initial WebSocket snapshots.
35 */
36 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
37}
38
39/**
40 * Merge an incoming full shell snapshot into prior client state.
41 *
42 * Authoritative snapshots (no options) replace structure and sequence even when
43 * lower than cache, while retaining a prior non-null identity when the candidate
44 * is still unresolved/null for the same root.
45 *
46 * Enrichment snapshots (options present) never resurrect or roll back
47 * projects/threads/archives/sequence when at or below cache; they only patch
48 * repository identity for matching current projects.
49 */
50export function mergeShellSnapshotProjects(
51 previous: OrchestrationV2ShellSnapshot | null | undefined,
52 next: OrchestrationV2ShellSnapshot,
53 options?: MergeShellSnapshotOptions,
54): OrchestrationV2ShellSnapshot {
55 if (previous === null || previous === undefined) {
56 return next;
57 }
58
59 const isEnrichment = options !== undefined;
60 const resolvedRootSet = isEnrichment ? new Set(options.resolvedRepositoryIdentityRoots) : null;
61
62 if (isEnrichment && next.snapshotSequence <= previous.snapshotSequence) {
63 const nextById = new Map(next.projects.map((project) => [project.id, project] as const));
64 return {
65 ...previous,
66 projects: previous.projects.map((project) => {
67 const candidate = nextById.get(project.id);
68 if (candidate === undefined || candidate.workspaceRoot !== project.workspaceRoot) {
69 return project;
70 }
71 if (resolvedRootSet?.has(project.workspaceRoot) === true) {
72 return { ...project, repositoryIdentity: candidate.repositoryIdentity };
73 }
74 if (project.repositoryIdentity == null && candidate.repositoryIdentity != null) {
75 return { ...project, repositoryIdentity: candidate.repositoryIdentity };
76 }
77 return project;
78 }),
79 };
80 }
81
82 const previousById = new Map(previous.projects.map((project) => [project.id, project] as const));
83 return {
84 ...next,
85 projects: next.projects.map((project) => {
86 const prior = previousById.get(project.id);
87 if (resolvedRootSet?.has(project.workspaceRoot) === true) {
88 return project;
89 }
90 return retainRepositoryIdentity(prior, project);
91 }),
92 };
93}
94
15/** Applies one committed V2 shell delta while preserving active/archive exclusivity. */95/** Applies one committed V2 shell delta while preserving active/archive exclusivity. */
16export function applyShellStreamEvent(96export function applyShellStreamEvent(
17 snapshot: OrchestrationV2ShellSnapshot,97 snapshot: OrchestrationV2ShellSnapshot,
@@ -20,12 +100,19 @@ export function applyShellStreamEvent( · 2 unmodified lines
20 if (event.sequence <= snapshot.snapshotSequence) return snapshot;100 if (event.sequence <= snapshot.snapshotSequence) return snapshot;
21 101
22 switch (event.kind) {102 switch (event.kind) {
23 case "project.updated":103 case "project.updated": {
104 // Enrichment is async. A project mutation can land with null
105 // repositoryIdentity while an earlier snapshot already resolved it.
106 // Keep the prior identity for the same workspace root so multi-env
107 // grouping does not split until a full snapshot refresh arrives.
108 const previous = snapshot.projects.find((project) => project.id === event.project.id);
109 const project = retainRepositoryIdentity(previous, event.project);
24 return {110 return {
25 ...snapshot,111 ...snapshot,
26 projects: upsertById(snapshot.projects, event.project),112 projects: upsertById(snapshot.projects, project),
27 snapshotSequence: event.sequence,113 snapshotSequence: event.sequence,
28 };114 };
115 }
29 case "project.removed":116 case "project.removed":
30 return {117 return {
31 ...snapshot,118 ...snapshot,

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t