import { assert, it, describe } from "@effect/vitest";import * as NodeServices from "@effect/platform-node/NodeServices";import * as Cause from "effect/Cause";import * as Deferred from "effect/Deferred";import * as DateTime from "effect/DateTime";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as Exit from "effect/Exit";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Logger from "effect/Logger";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock";import type {  BackgroundScope,  VcsStatusLocalResult,  VcsStatusRemoteResult,  VcsStatusResult,  VcsStatusStreamEvent,} from "@t3tools/contracts";import { GitManagerError } from "@t3tools/contracts"; import * as VcsStatusBroadcaster from "./VcsStatusBroadcaster.ts";import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";import * as GitWorkflowService from "../git/GitWorkflowService.ts";import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z"); const baseLocalStatus: VcsStatusLocalResult = {  isRepo: true,  sourceControlProvider: {    kind: "github",    name: "GitHub",    baseUrl: "https://github.com",  },  hasPrimaryRemote: true,  isDefaultRef: false,  refName: "feature/status-broadcast",  hasWorkingTreeChanges: false,  workingTree: { files: [], insertions: 0, deletions: 0 },}; const baseRemoteStatus: VcsStatusRemoteResult = {  hasUpstream: true,  aheadCount: 0,  behindCount: 0,  pr: null,}; const remoteStatusWithPr: VcsStatusRemoteResult = {  ...baseRemoteStatus,  pr: {    number: 2978,    title: "[codex] Rewrite client connection architecture",    url: "https://github.com/pingdotgg/t3code/pull/2978",    baseRef: "main",    headRef: "codex/connection-state-audit",    state: "open",  },}; const baseStatus: VcsStatusResult = {  ...baseLocalStatus,  ...baseRemoteStatus,}; function makeTestLayer(state: {  currentLocalStatus: VcsStatusLocalResult;  currentRemoteStatus: VcsStatusRemoteResult | null;  localStatusCalls: number;  remoteStatusCalls: number;  localInvalidationCalls: number;  remoteInvalidationCalls: number;  remoteStatusRefreshUpstreamValues?: Array<boolean | undefined>;  backgroundWorkEnabled?: boolean;}) {  return VcsStatusBroadcaster.layer.pipe(    Layer.provideMerge(NodeServices.layer),    Layer.provide(makeBackgroundPolicyLayer(() => state.backgroundWorkEnabled !== false)),    Layer.provide(      Layer.mock(GitWorkflowService.GitWorkflowService)({        localStatus: () =>          Effect.sync(() => {            state.localStatusCalls += 1;            return state.currentLocalStatus;          }),        remoteStatus: (_input, options) =>          Effect.sync(() => {            state.remoteStatusCalls += 1;            state.remoteStatusRefreshUpstreamValues?.push(options?.refreshUpstream);            return state.currentRemoteStatus;          }),        invalidateLocalStatus: () =>          Effect.sync(() => {            state.localInvalidationCalls += 1;          }),        invalidateRemoteStatus: () =>          Effect.sync(() => {            state.remoteInvalidationCalls += 1;          }),        invalidateStatus: () =>          Effect.sync(() => {            state.localInvalidationCalls += 1;            state.remoteInvalidationCalls += 1;          }),      }),    ),  );} function makeBackgroundPolicyLayer(shouldRunScopeWork: (scope: BackgroundScope) => boolean) {  return Layer.mock(BackgroundPolicy.BackgroundPolicy)({    reportClientActivity: () => Effect.void,    removeRpcClient: () => Effect.void,    reportHostPowerState: () => Effect.void,    snapshot: Effect.succeed({      hostPower: {        source: "unknown",        idle: "unknown",        idleSeconds: null,        locked: "unknown",        suspended: false,        onBattery: "unknown",        lowPowerMode: "unknown",        thermalState: "unknown",        stale: true,        updatedAt: TEST_EPOCH,      },      leases: [],      activeForegroundLeaseCount: 0,      activeScopeKeys: [],      shouldRunOpportunisticWork: false,      updatedAt: TEST_EPOCH,    }),    streamChanges: Stream.empty,    hasDemand: () => Effect.succeed(true),    shouldRunScopeWork: (scope) => Effect.sync(() => shouldRunScopeWork(scope)),    shouldRunOpportunisticWork: Effect.succeed(true),  });} describe("VcsStatusBroadcaster", () => {  it.effect.skipIf(!symlinksSupported)(    "automatically pulls an enabled clean default branch when status detects it is behind",    () => {      let remoteStatus: VcsStatusRemoteResult = { ...baseRemoteStatus, behindCount: 2 };      let pullCalls = 0;      let configuredWorkspaceRoot = "";      const localStatus: VcsStatusLocalResult = {        ...baseLocalStatus,        isDefaultRef: true,        refName: "main",      };      const testLayer = VcsStatusBroadcaster.layer.pipe(        Layer.provideMerge(NodeServices.layer),        Layer.provide(makeBackgroundPolicyLayer(() => true)),        Layer.provide(          Layer.succeed(VcsStatusBroadcaster.VcsAutoPullPolicy, {            isEnabled: (cwd) => Effect.succeed(cwd === configuredWorkspaceRoot),          }),        ),        Layer.provide(          Layer.mock(GitWorkflowService.GitWorkflowService)({            localStatus: () => Effect.succeed(localStatus),            remoteStatus: () => Effect.succeed(remoteStatus),            invalidateLocalStatus: () => Effect.void,            invalidateRemoteStatus: () => Effect.void,            invalidateStatus: () => Effect.void,            pullCurrentBranch: () =>              Effect.sync(() => {                pullCalls += 1;                remoteStatus = { ...remoteStatus, behindCount: 0 };                return {                  status: "pulled" as const,                  refName: "main",                  upstreamRef: "origin/main",                };              }),          }),        ),      );       return Effect.gen(function* () {        const fileSystem = yield* FileSystem.FileSystem;        const path = yield* Path.Path;        const realDir = yield* fileSystem.makeTempDirectoryScoped({          prefix: "t3-vcs-auto-pull-real-",        });        const linkParent = yield* fileSystem.makeTempDirectoryScoped({          prefix: "t3-vcs-auto-pull-link-",        });        configuredWorkspaceRoot = path.join(linkParent, "repo-link");        yield* fileSystem.symlink(realDir, configuredWorkspaceRoot);         const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;        const status = yield* broadcaster.refreshStatus(configuredWorkspaceRoot);         assert.equal(pullCalls, 1);        assert.equal(status.behindCount, 0);      }).pipe(Effect.provide(testLayer));    },  );   it.effect("reuses the cached VCS status across repeated reads", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;       const first = yield* broadcaster.getStatus({ cwd: "/repo" });      const second = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.deepStrictEqual(first, baseStatus);      assert.deepStrictEqual(second, baseStatus);      assert.equal(state.localStatusCalls, 1);      assert.equal(state.remoteStatusCalls, 1);      assert.equal(state.localInvalidationCalls, 0);      assert.equal(state.remoteInvalidationCalls, 0);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("refreshes a loaded cwd without reusing a previous branch's PR", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;       // Nobody loaded this cwd yet: no host request is spent.      assert.isNull(yield* broadcaster.refreshPullRequestStatus("/repo"));      assert.equal(state.remoteStatusCalls, 0);       yield* broadcaster.getStatus({ cwd: "/repo" });      assert.equal(state.remoteStatusCalls, 1);       // Loaded and no PR known: ask GitManager to retry the missing PR.      state.currentRemoteStatus = remoteStatusWithPr;      const refreshed = yield* broadcaster.refreshPullRequestStatus("/repo");      assert.deepStrictEqual(refreshed, remoteStatusWithPr);      assert.equal(state.remoteStatusCalls, 2);      assert.equal(state.remoteInvalidationCalls, 0);       // The agent switches branches. The previous branch's PR must not block a read.      state.currentLocalStatus = { ...baseLocalStatus, refName: "feature/next" };      state.currentRemoteStatus = baseRemoteStatus;      yield* broadcaster.refreshLocalStatus("/repo");      const refreshedBranch = yield* broadcaster.refreshPullRequestStatus("/repo");      assert.deepStrictEqual(refreshedBranch, baseRemoteStatus);      assert.equal(state.remoteStatusCalls, 3);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("a poll that started before the turn-end refresh cannot overwrite its PR", () => {    const releaseFirstPoll = Deferred.makeUnsafe<void>();    const firstPollStarted = Deferred.makeUnsafe<void>();    let remoteReads = 0;    const layer = VcsStatusBroadcaster.layer.pipe(      Layer.provideMerge(NodeServices.layer),      Layer.provide(makeBackgroundPolicyLayer(() => true)),      Layer.provide(        Layer.mock(GitWorkflowService.GitWorkflowService)({          localStatus: () => Effect.succeed(baseLocalStatus),          remoteStatus: () =>            Effect.gen(function* () {              remoteReads += 1;              if (remoteReads === 2) {                // Hold an older empty response while the turn-end refresh queues.                yield* Deferred.succeed(firstPollStarted, undefined);                yield* Deferred.await(releaseFirstPoll);                return baseRemoteStatus;              }              return remoteReads === 1 ? baseRemoteStatus : remoteStatusWithPr;            }),          invalidateLocalStatus: () => Effect.void,          invalidateRemoteStatus: () => Effect.void,          invalidateStatus: () => Effect.void,        }),      ),    );     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      yield* broadcaster.getStatus({ cwd: "/repo" });       const poll = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);      yield* Deferred.await(firstPollStarted);      const refresh = yield* broadcaster.refreshPullRequestStatus("/repo").pipe(Effect.forkScoped);      yield* Deferred.succeed(releaseFirstPoll, undefined);      yield* Fiber.join(poll);      const refreshed = yield* Fiber.join(refresh);       assert.deepStrictEqual(refreshed, remoteStatusWithPr);      const final = yield* broadcaster.getStatus({ cwd: "/repo" });      assert.deepStrictEqual(final.pr, remoteStatusWithPr.pr);    }).pipe(Effect.provide(layer), Effect.scoped);  });   it.effect("an initial status read cannot overwrite an explicit refresh", () => {    const firstReadStarted = Deferred.makeUnsafe<void>();    const releaseFirstRead = Deferred.makeUnsafe<void>();    let remoteReads = 0;    const layer = VcsStatusBroadcaster.layer.pipe(      Layer.provide(FileSystem.layerNoop({ realPath: (path) => Effect.succeed(path) })),      Layer.provideMerge(NodeServices.layer),      Layer.provide(makeBackgroundPolicyLayer(() => true)),      Layer.provide(        Layer.mock(GitWorkflowService.GitWorkflowService)({          localStatus: () => Effect.succeed(baseLocalStatus),          remoteStatus: () =>            Effect.gen(function* () {              remoteReads += 1;              if (remoteReads === 1) {                yield* Deferred.succeed(firstReadStarted, undefined);                yield* Deferred.await(releaseFirstRead);                return baseRemoteStatus;              }              return remoteStatusWithPr;            }),          invalidateStatus: () => Effect.void,        }),      ),    );    return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const initial = yield* broadcaster.getStatus({ cwd: "/repo" }).pipe(Effect.forkScoped);      yield* Deferred.await(firstReadStarted);      const refresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);      // Run ready fibers before releasing the delayed first read.      yield* TestClock.adjust(Duration.zero);      yield* Deferred.succeed(releaseFirstRead, undefined);      yield* Fiber.join(initial);      yield* Fiber.join(refresh);      assert.deepStrictEqual(        (yield* broadcaster.getStatus({ cwd: "/repo" })).pr,        remoteStatusWithPr.pr,      );    }).pipe(Effect.provide(layer), Effect.scoped);  });   it.effect("turn-end refresh skips a loaded cwd when background policy pauses it", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,      backgroundWorkEnabled: false,    };    return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      yield* broadcaster.getStatus({ cwd: "/repo" });      yield* broadcaster.refreshPullRequestStatus("/repo");      assert.equal(state.remoteStatusCalls, 1);      assert.equal(state.remoteInvalidationCalls, 0);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("refreshes the cached snapshot after explicit invalidation", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const initial = yield* broadcaster.getStatus({ cwd: "/repo" });       state.currentLocalStatus = {        ...baseLocalStatus,        refName: "feature/updated-status",      };      state.currentRemoteStatus = {        ...baseRemoteStatus,        aheadCount: 2,      };      const refreshed = yield* broadcaster.refreshStatus("/repo");      const cached = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.deepStrictEqual(initial, baseStatus);      assert.deepStrictEqual(refreshed, {        ...state.currentLocalStatus,        ...state.currentRemoteStatus,      });      assert.deepStrictEqual(cached, {        ...state.currentLocalStatus,        ...state.currentRemoteStatus,      });      assert.equal(state.localStatusCalls, 2);      assert.equal(state.remoteStatusCalls, 2);      assert.equal(state.localInvalidationCalls, 1);      assert.equal(state.remoteInvalidationCalls, 1);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("keeps the cached snapshot unchanged when a refresh branch fails", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,      failRemoteStatus: false,    };    const testLayer = VcsStatusBroadcaster.layer.pipe(      Layer.provideMerge(NodeServices.layer),      Layer.provide(makeBackgroundPolicyLayer(() => true)),      Layer.provide(        Layer.mock(GitWorkflowService.GitWorkflowService)({          localStatus: () =>            Effect.sync(() => {              state.localStatusCalls += 1;              return state.currentLocalStatus;            }),          remoteStatus: () =>            Effect.suspend(() => {              state.remoteStatusCalls += 1;              return state.failRemoteStatus                ? Effect.fail(                    new GitManagerError({                      operation: "VcsStatusBroadcaster.test",                      cwd: "/repo",                      detail: "remote status failed",                    }),                  )                : Effect.succeed(state.currentRemoteStatus);            }),          invalidateLocalStatus: () =>            Effect.sync(() => {              state.localInvalidationCalls += 1;            }),          invalidateRemoteStatus: () =>            Effect.sync(() => {              state.remoteInvalidationCalls += 1;            }),          invalidateStatus: () =>            Effect.sync(() => {              state.localInvalidationCalls += 1;              state.remoteInvalidationCalls += 1;            }),        }),      ),    );     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      yield* broadcaster.getStatus({ cwd: "/repo" });       state.currentLocalStatus = {        ...baseLocalStatus,        refName: "feature/partial-refresh",      };      state.currentRemoteStatus = {        ...baseRemoteStatus,        aheadCount: 3,      };      state.failRemoteStatus = true;       const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit);      const cached = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.isTrue(Exit.isFailure(refreshExit));      assert.deepStrictEqual(cached, baseStatus);    }).pipe(Effect.provide(testLayer));  });   it.effect("refreshes only the cached local snapshot when requested", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {

Lines 1–500 of 1030. Load more