import * as Cause from "effect/Cause";import * as Context from "effect/Context";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as PubSub from "effect/PubSub";import * as Ref from "effect/Ref";import * as Schedule from "effect/Schedule";import * as Scope from "effect/Scope";import * as Semaphore from "effect/Semaphore";import * as Stream from "effect/Stream";import * as SynchronizedRef from "effect/SynchronizedRef";import type {  GitManagerServiceError,  VcsStatusInput,  VcsStatusLocalResult,  VcsStatusRemoteResult,  VcsStatusResult,  VcsStatusStreamEvent,} from "@t3tools/contracts";import { mergeGitStatusParts } from "@t3tools/shared/git";import { resolveProjectAutoPull } from "@t3tools/shared/serverSettings"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";import * as GitWorkflowService from "../git/GitWorkflowService.ts";import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts";import * as ServerSettings from "../serverSettings.ts"; const DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30);const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30);const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15);const MAX_FAILURE_DIAGNOSTIC_VALUES = 8;const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128; function boundedDiagnosticValue(value: string): string {  return value.slice(0, MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH);} function diagnosticValueTag(value: unknown): string {  try {    if (      typeof value === "object" &&      value !== null &&      "_tag" in value &&      typeof value._tag === "string"    ) {      return boundedDiagnosticValue(value._tag);    }    if (value instanceof Error) {      return boundedDiagnosticValue(value.name);    }    return typeof value;  } catch {    return "Uninspectable";  }} function diagnosticFailureOperation(value: unknown): string | undefined {  try {    if (      typeof value === "object" &&      value !== null &&      "operation" in value &&      typeof value.operation === "string"    ) {      return boundedDiagnosticValue(value.operation);    }  } catch {    return undefined;  }  return undefined;} function addUniqueDiagnosticValue(values: Array<string>, value: string | undefined): void {  if (    value !== undefined &&    values.length < MAX_FAILURE_DIAGNOSTIC_VALUES &&    !values.includes(value)  ) {    values.push(value);  }} export function remoteRefreshFailureDiagnostics(cause: Cause.Cause<unknown>) {  const failureTags: Array<string> = [];  const failureOperations: Array<string> = [];  const defectTags: Array<string> = [];  let failureCount = 0;  let defectCount = 0;  let interruptionCount = 0;   for (const reason of cause.reasons) {    if (Cause.isFailReason(reason)) {      failureCount += 1;      addUniqueDiagnosticValue(failureTags, diagnosticValueTag(reason.error));      addUniqueDiagnosticValue(failureOperations, diagnosticFailureOperation(reason.error));      continue;    }    if (Cause.isDieReason(reason)) {      defectCount += 1;      addUniqueDiagnosticValue(defectTags, diagnosticValueTag(reason.defect));      continue;    }    interruptionCount += 1;  }   return {    reasonCount: cause.reasons.length,    failureCount,    failureTags,    failureOperations,    defectCount,    defectTags,    interruptionCount,  };} interface VcsStatusChange {  readonly cwd: string;  readonly event: VcsStatusStreamEvent;} interface CachedValue<T> {  readonly fingerprint: string;  readonly value: T;} interface CachedVcsStatus {  readonly local: CachedValue<VcsStatusLocalResult> | null;  readonly remote: CachedValue<VcsStatusRemoteResult | null> | null;} interface ActiveRemotePoller {  readonly fiber: Fiber.Fiber<void, never>;  readonly subscriberCount: number;  readonly demandCwds: Ref.Ref<ReadonlyMap<string, number>>;} interface StreamStatusOptions {  readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>;} export class VcsAutoPullPolicy extends Context.Reference<{  readonly isEnabled: (cwd: string) => Effect.Effect<boolean, never>;}>("t3/vcs/VcsAutoPullPolicy", {  defaultValue: () => ({ isEnabled: () => Effect.succeed(false) }),}) {} export const autoPullPolicyLayer = Layer.effect(  VcsAutoPullPolicy,  Effect.gen(function* () {    const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;    const serverSettings = yield* ServerSettings.ServerSettingsService;    return {      isEnabled: Effect.fn("VcsAutoPullPolicy.isEnabled")(        function* (cwd: string) {          const project = yield* snapshots.getActiveProjectByWorkspaceRoot(cwd);          if (project._tag === "None") return false;          const settings = yield* serverSettings.getSettings;          return resolveProjectAutoPull(settings, project.value.id, project.value.autoPull);        },        Effect.orElseSucceed(() => false),      ),    };  }),); export function remoteRefreshFailureDelay(  consecutiveFailures: number,  configuredInterval: Duration.Duration,) {  const exponent = Math.max(0, consecutiveFailures - 1);  const backoffMs =    Duration.toMillis(VCS_STATUS_REFRESH_FAILURE_BASE_DELAY) * Math.pow(2, exponent);  const cappedBackoff = Duration.min(    Duration.millis(backoffMs),    VCS_STATUS_REFRESH_FAILURE_MAX_DELAY,  );  return Duration.max(configuredInterval, cappedBackoff);} export class VcsStatusBroadcaster extends Context.Service<  VcsStatusBroadcaster,  {    readonly getStatus: (      input: VcsStatusInput,    ) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;    readonly refreshLocalStatus: (      cwd: string,    ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>;    readonly refreshStatus: (cwd: string) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;    /**     * Refresh a loaded cwd after a turn if background policy allows it.     * GitManager retries missing PRs for the current branch and keeps known     * PRs and failed lookup backoff cached. This does not fetch Git remotes.     */    readonly refreshPullRequestStatus: (      cwd: string,    ) => Effect.Effect<VcsStatusRemoteResult | null, GitManagerServiceError>;    readonly streamStatus: (      input: VcsStatusInput,      options?: StreamStatusOptions,    ) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>;  }>()("t3/vcs/VcsStatusBroadcaster") {} function fingerprintStatusPart(status: unknown): string {  return JSON.stringify(status);} const normalizeCwd = (cwd: string) =>  Effect.service(FileSystem.FileSystem).pipe(    Effect.flatMap((fs) => fs.realPath(cwd)),    Effect.orElseSucceed(() => cwd),  ); /** @public Service construction is part of the canonical Effect module API. */export const make = Effect.gen(function* () {  const autoPullPolicy = yield* VcsAutoPullPolicy;  const workflow = yield* GitWorkflowService.GitWorkflowService;  const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;  const fs = yield* FileSystem.FileSystem;  const changesPubSub = yield* Effect.acquireRelease(    PubSub.unbounded<VcsStatusChange>(),    (pubsub) => PubSub.shutdown(pubsub),  );  const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>    Scope.close(scope, Exit.void),  );  const cacheRef = yield* Ref.make(new Map<string, CachedVcsStatus>());  // One permit per cwd for remote reads that write the cache. Without it a  // periodic poll that started before `gh pr create` can finish after the  // turn-end refresh and overwrite the fresh PR with its stale `pr: null`.  const remoteWriteLocks = new Map<string, Semaphore.Semaphore>();  const withRemoteWriteLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) => {    let lock = remoteWriteLocks.get(cwd);    if (lock === undefined) {      lock = Semaphore.makeUnsafe(1);      remoteWriteLocks.set(cwd, lock);    }    return lock.withPermits(1)(effect);  };  const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>());   const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* (    cwd: string,  ) {    return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null));  });   const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")(    function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) {      const nextLocal = {        fingerprint: fingerprintStatusPart(local),        value: local,      } satisfies CachedValue<VcsStatusLocalResult>;      const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {        const previous = cache.get(cwd) ?? { local: null, remote: null };        const nextCache = new Map(cache);        nextCache.set(cwd, {          ...previous,          local: nextLocal,        });        return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const;      });       if (options?.publish && shouldPublish) {        yield* PubSub.publish(changesPubSub, {          cwd,          event: {            _tag: "localUpdated",            local,          },        });      }       return local;    },  );   const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")(    function* (cwd: string, remote: VcsStatusRemoteResult | null, options?: { publish?: boolean }) {      const nextRemote = {        fingerprint: fingerprintStatusPart(remote),        value: remote,      } satisfies CachedValue<VcsStatusRemoteResult | null>;      const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {        const previous = cache.get(cwd) ?? { local: null, remote: null };        const nextCache = new Map(cache);        nextCache.set(cwd, {          ...previous,          remote: nextRemote,        });        return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const;      });       if (options?.publish && shouldPublish) {        yield* PubSub.publish(changesPubSub, {          cwd,          event: {            _tag: "remoteUpdated",            remote,          },        });      }       return remote;    },  );   const updateCachedStatus = Effect.fn("VcsStatusBroadcaster.updateCachedStatus")(function* (    cwd: string,    local: VcsStatusLocalResult,    remote: VcsStatusRemoteResult | null,    options?: { publish?: boolean },  ) {    const nextLocal = {      fingerprint: fingerprintStatusPart(local),      value: local,    } satisfies CachedValue<VcsStatusLocalResult>;    const nextRemote = {      fingerprint: fingerprintStatusPart(remote),      value: remote,    } satisfies CachedValue<VcsStatusRemoteResult | null>;    const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {      const previous = cache.get(cwd) ?? { local: null, remote: null };      const nextCache = new Map(cache);      nextCache.set(cwd, {        local: nextLocal,        remote: nextRemote,      });      return [        previous.local?.fingerprint !== nextLocal.fingerprint ||          previous.remote?.fingerprint !== nextRemote.fingerprint,        nextCache,      ] as const;    });     if (options?.publish && shouldPublish) {      yield* PubSub.publish(changesPubSub, {        cwd,        event: {          _tag: "snapshot",          local,          remote,        },      });    }     return mergeGitStatusParts(local, remote);  });   const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (    cwd: string,  ) {    const local = yield* workflow.localStatus({ cwd });    return yield* updateCachedLocalStatus(cwd, local);  });   const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* (    cwd: string,  ) {    const cached = yield* getCachedStatus(cwd);    if (cached?.local) {      return cached.local.value;    }    return yield* loadLocalStatus(cwd);  });   const withFileSystem = Effect.provideService(FileSystem.FileSystem, fs);   const getStatus: VcsStatusBroadcaster["Service"]["getStatus"] = Effect.fn(    "VcsStatusBroadcaster.getStatus",  )(function* (input) {    const cwd = yield* withFileSystem(normalizeCwd(input.cwd));    const cached = yield* getCachedStatus(cwd);    if (cached?.local && cached.remote) {      return mergeGitStatusParts(cached.local.value, cached.remote.value);    }    return yield* withRemoteWriteLock(      cwd,      Effect.gen(function* () {        const latest = yield* getCachedStatus(cwd);        const [local, remote] = yield* Effect.all(          [            latest?.local ? Effect.succeed(latest.local.value) : workflow.localStatus({ cwd }),            latest?.remote ? Effect.succeed(latest.remote.value) : workflow.remoteStatus({ cwd }),          ],          { concurrency: "unbounded" },        );        return yield* updateCachedStatus(cwd, local, remote);      }),    );  });   const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")(    function* (cwd: string) {      yield* workflow.invalidateLocalStatus(cwd);      const local = yield* workflow.localStatus({ cwd });      return yield* updateCachedLocalStatus(cwd, local, { publish: true });    },  );   const refreshLocalStatus: VcsStatusBroadcaster["Service"]["refreshLocalStatus"] = Effect.fn(    "VcsStatusBroadcaster.refreshLocalStatus",  )(function* (rawCwd) {    const cwd = yield* withFileSystem(normalizeCwd(rawCwd));    return yield* refreshLocalStatusCore(cwd);  });   const maybeAutoPull = Effect.fn("VcsStatusBroadcaster.maybeAutoPull")(function* (    cwd: string,    remote: VcsStatusRemoteResult | null,    policyCwds: ReadonlyArray<string>,  ) {    return yield* Effect.gen(function* () {      const autoPullEnabled = (yield* Effect.forEach(policyCwds, autoPullPolicy.isEnabled, {        concurrency: "unbounded",      })).some(Boolean);      if (        remote === null ||        !remote.hasUpstream ||        remote.aheadCount > 0 ||        remote.behindCount <= 0 ||        !autoPullEnabled      ) {        return null;      }       yield* workflow.invalidateLocalStatus(cwd);      const local = yield* workflow.localStatus({ cwd });      if (!local.isRepo || !local.isDefaultRef || local.hasWorkingTreeChanges) return null;       yield* workflow.pullCurrentBranch(cwd);      yield* workflow.invalidateStatus(cwd);      const [refreshedLocal, refreshedRemote] = yield* Effect.all(        [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd }, { refreshUpstream: false })],        { concurrency: "unbounded" },      );      yield* updateCachedStatus(cwd, refreshedLocal, refreshedRemote, { publish: true });      return { local: refreshedLocal, remote: refreshedRemote };    }).pipe(      Effect.catch(() =>        Effect.logWarning("Automatic project pull failed", { cwd }).pipe(Effect.as(null)),      ),    );  });   const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (    cwd: string,    options?: {      readonly refreshUpstream?: boolean;      readonly policyCwds?: ReadonlyArray<string>;    },  ) {    return yield* withRemoteWriteLock(      cwd,      Effect.gen(function* () {        if (options?.refreshUpstream !== false) {          yield* workflow.invalidateRemoteStatus(cwd);        }        const remote = yield* workflow.remoteStatus({ cwd }, options);        const pulled = yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]);        if (pulled !== null) return pulled.remote;        return yield* updateCachedRemoteStatus(cwd, remote, { publish: true });      }),    );  });   const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn(    "VcsStatusBroadcaster.refreshStatus",  )(function* (rawCwd) {    const cwd = yield* withFileSystem(normalizeCwd(rawCwd));    // invalidateStatus (not the two partial invalidations) so an explicit    // refresh also bypasses GitManager's slow PR-lookup cache.    return yield* withRemoteWriteLock(      cwd,      Effect.gen(function* () {        yield* workflow.invalidateStatus(cwd);        const [local, remote] = yield* Effect.all(          [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })],          { concurrency: "unbounded" },        );        const pulled = yield* maybeAutoPull(cwd, remote, [rawCwd]);        if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote);        return yield* updateCachedStatus(cwd, local, remote, { publish: true });      }),    );  });   const refreshPullRequestStatus: VcsStatusBroadcaster["Service"]["refreshPullRequestStatus"] =    Effect.fn("VcsStatusBroadcaster.refreshPullRequestStatus")(function* (rawCwd) {      const cwd = yield* withFileSystem(normalizeCwd(rawCwd));      return yield* withRemoteWriteLock(        cwd,        Effect.gen(function* () {          const cached = yield* getCachedStatus(cwd);

Lines 1–500 of 738. Load more