Add Pi provider integration and improve git status handling

apps/server/integration/orchestrationEngine.integration.test.ts

@@ -109,6 +109,9 @@ const seedProjectAndThread = (harness: OrchestrationIntegrationHarness) =>
   Effect.gen(function* () {
     const createdAt = nowIso();
     const provider = harness.adapterHarness?.provider ?? "codex";
+    if (provider === "pi") {
+      throw new Error("Pi integration tests require an explicit model selection.");
+    }
     const defaultModel = DEFAULT_MODEL_BY_PROVIDER[provider];
 
     yield* harness.engine.dispatch({

apps/server/package.json

@@ -24,6 +24,9 @@
   },
   "dependencies": {
     "@anthropic-ai/claude-agent-sdk": "^0.2.111",
+    "@earendil-works/pi-agent-core": "^0.74.0",
+    "@earendil-works/pi-ai": "^0.74.0",
+    "@earendil-works/pi-coding-agent": "^0.74.0",
     "@effect/platform-node": "catalog:",
     "@effect/sql-sqlite-bun": "catalog:",
     "@opencode-ai/sdk": "^1.14.48",

apps/server/src/git/Layers/GitCore.test.ts

@@ -2043,6 +2043,39 @@ it.layer(TestLayer)("git integration", (it) => {
       }),
     );
 
+    it.effect("explains local changes that block pull", () =>
+      Effect.gen(function* () {
+        const remote = yield* makeTmpDir();
+        const source = yield* makeTmpDir();
+        const clone = yield* makeTmpDir();
+        yield* git(remote, ["init", "--bare"]);
+
+        yield* initRepoWithCommit(source);
+        const initialBranch = (yield* (yield* GitCore).listBranches({ cwd: source })).branches.find(
+          (branch) => branch.current,
+        )!.name;
+        yield* git(source, ["remote", "add", "origin", remote]);
+        yield* git(source, ["push", "-u", "origin", initialBranch]);
+
+        yield* git(clone, ["clone", remote, "."]);
+        yield* git(clone, ["config", "user.email", "test@test.com"]);
+        yield* git(clone, ["config", "user.name", "Test"]);
+        yield* writeTextFile(path.join(clone, "README.md"), "remote change\n");
+        yield* git(clone, ["add", "README.md"]);
+        yield* git(clone, ["commit", "-m", "remote update"]);
+        yield* git(clone, ["push", "origin", initialBranch]);
+
+        yield* writeTextFile(path.join(source, "README.md"), "local change\n");
+
+        const result = yield* Effect.result((yield* GitCore).pullCurrentBranch(source));
+        expect(result._tag).toBe("Failure");
+        if (result._tag === "Failure") {
+          expect(result.failure.detail).toContain("Local changes block pull");
+          expect(result.failure.detail).toContain("README.md");
+        }
+      }),
+    );
+
     it.effect("lists branches when recency lookup fails", () =>
       Effect.gen(function* () {
         const tmp = yield* makeTmpDir();

apps/server/src/git/Layers/GitCore.ts

@@ -376,7 +376,7 @@ function createGitCommandError(
 const DIRTY_WORKTREE_PATTERN =
   /Your local changes to the following files would be overwritten by (?:checkout|merge):\s*([\s\S]*?)Please commit your changes or stash them/;
 const UNTRACKED_OVERWRITE_PATTERN =
-  /The following untracked working tree files would be overwritten by checkout:\s*([\s\S]*?)Please move or remove them/;
+  /The following untracked working tree files would be overwritten by (?:checkout|merge):\s*([\s\S]*?)Please move or remove them/;
 
 function parseDirtyWorktreeFiles(stderr: string): string[] | null {
   const match = DIRTY_WORKTREE_PATTERN.exec(stderr) ?? UNTRACKED_OVERWRITE_PATTERN.exec(stderr);
@@ -388,6 +388,13 @@ function parseDirtyWorktreeFiles(stderr: string): string[] | null {
   return files.length > 0 ? files : null;
 }
 
+function explainPullBlockedByLocalChanges(error: GitCommandError): string | null {
+  const files = parseDirtyWorktreeFiles(error.detail);
+  if (!files) return null;
+  const fileList = files.map((file) => `  - ${file}`).join("\n");
+  return `Local changes block pull. Commit or stash these files first:\n${fileList}`;
+}
+
 function parseNonEmptyLineList(input: string): string[] {
   return input
     .trim()
@@ -1709,7 +1716,19 @@ export const makeGitCore = (options?: { executeOverride?: GitCoreShape["execute"
         yield* executeGit("GitCore.pullCurrentBranch.pull", cwd, ["pull", "--ff-only"], {
           timeoutMs: 30_000,
           fallbackErrorMessage: "git pull failed",
-        });
+        }).pipe(
+          Effect.mapError((error) => {
+            const friendlyDetail = explainPullBlockedByLocalChanges(error);
+            if (!friendlyDetail) return error;
+            return createGitCommandError(
+              "GitCore.pullCurrentBranch.pull",
+              cwd,
+              ["pull", "--ff-only"],
+              friendlyDetail,
+              error,
+            );
+          }),
+        );
         const afterSha = yield* runGitStdout(
           "GitCore.pullCurrentBranch.afterSha",
           cwd,

apps/server/src/git/Layers/GitStatusBroadcaster.test.ts

@@ -1,8 +1,9 @@
 import type { GitStatusResult, GitStatusStreamEvent } from "@t3tools/contracts";
 import { Deferred, Effect, Layer, Scope, Stream } from "effect";
-import { describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 
 import type { GitManagerServiceError } from "../Errors";
+import { GitCore, type GitCoreShape, type GitStatusDetails } from "../Services/GitCore";
 import { GitManager, type GitManagerShape } from "../Services/GitManager";
 import { GitStatusBroadcaster } from "../Services/GitStatusBroadcaster";
 import { GitStatusBroadcasterLive } from "./GitStatusBroadcaster";
@@ -17,7 +18,29 @@ const baseStatus: GitStatusResult = {
   pr: null,
 };
 
-function makeTestLayer(state: { currentStatus: GitStatusResult; statusCalls: number }) {
+const baseDetails: GitStatusDetails = {
+  branch: baseStatus.branch,
+  upstreamRef: "origin/feature/status-broadcast",
+  hasWorkingTreeChanges: baseStatus.hasWorkingTreeChanges,
+  workingTree: baseStatus.workingTree,
+  hasUpstream: baseStatus.hasUpstream,
+  aheadCount: baseStatus.aheadCount,
+  behindCount: baseStatus.behindCount,
+};
+
+function makeTestLayer(state: {
+  currentDetails: GitStatusDetails;
+  currentStatus: GitStatusResult;
+  detailsCalls: number;
+  statusCalls: number;
+}) {
+  const gitCore = {
+    statusDetails: () =>
+      Effect.sync(() => {
+        state.detailsCalls += 1;
+        return state.currentDetails;
+      }),
+  } as unknown as GitCoreShape;
   const gitManager: GitManagerShape = {
     status: () =>
       Effect.sync(() => {
@@ -33,35 +56,146 @@ function makeTestLayer(state: { currentStatus: GitStatusResult; statusCalls: num
     runStackedAction: () => Effect.die("runStackedAction should not be called in this test"),
   };
 
-  return GitStatusBroadcasterLive.pipe(Layer.provide(Layer.succeed(GitManager, gitManager)));
+  return GitStatusBroadcasterLive.pipe(
+    Layer.provide(
+      Layer.mergeAll(Layer.succeed(GitCore, gitCore), Layer.succeed(GitManager, gitManager)),
+    ),
+  );
 }
 
 const runBroadcasterTest = (
-  state: { currentStatus: GitStatusResult; statusCalls: number },
+  state: {
+    currentDetails: GitStatusDetails;
+    currentStatus: GitStatusResult;
+    detailsCalls: number;
+    statusCalls: number;
+  },
   effect: Effect.Effect<void, GitManagerServiceError, GitStatusBroadcaster | Scope.Scope>,
 ) => effect.pipe(Effect.provide(makeTestLayer(state)), Effect.scoped, Effect.runPromise);
 
+afterEach(() => {
+  vi.useRealTimers();
+});
+
 describe("GitStatusBroadcasterLive", () => {
-  it("reuses the cached git status across repeated reads", async () => {
-    const state = { currentStatus: baseStatus, statusCalls: 0 };
+  it("refreshes local git status on repeated reads without repeating PR lookup", async () => {
+    const state = {
+      currentDetails: baseDetails,
+      currentStatus: baseStatus,
+      detailsCalls: 0,
+      statusCalls: 0,
+    };
 
     await runBroadcasterTest(
       state,
       Effect.gen(function* () {
         const broadcaster = yield* GitStatusBroadcaster;
 
         const first = yield* broadcaster.getStatus({ cwd: "/repo" });
+        state.currentDetails = {
+          ...baseDetails,
+          hasWorkingTreeChanges: true,
+          workingTree: {
+            files: [{ path: "src/app.ts", insertions: 5, deletions: 1 }],
+            insertions: 5,
+            deletions: 1,
+          },
+        };
         const second = yield* broadcaster.getStatus({ cwd: "/repo" });
 
         expect(first).toEqual(baseStatus);
-        expect(second).toEqual(baseStatus);
+        expect(second).toEqual({
+          ...baseStatus,
+          hasWorkingTreeChanges: true,
+          workingTree: state.currentDetails.workingTree,
+        });
         expect(state.statusCalls).toBe(1);
+        expect(state.detailsCalls).toBe(1);
+      }),
+    );
+  });
+
+  it("refreshes full status when cached remote metadata expires", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(0);
+    const state = {
+      currentDetails: baseDetails,
+      currentStatus: baseStatus,
+      detailsCalls: 0,
+      statusCalls: 0,
+    };
+
+    await runBroadcasterTest(
+      state,
+      Effect.gen(function* () {
+        const broadcaster = yield* GitStatusBroadcaster;
+
+        const first = yield* broadcaster.getStatus({ cwd: "/repo" });
+        vi.setSystemTime(31_000);
+        state.currentStatus = {
+          ...baseStatus,
+          pr: {
+            number: 42,
+            title: "Open PR",
+            url: "https://github.com/acme/repo/pull/42",
+            state: "open",
+          },
+        };
+        const second = yield* broadcaster.getStatus({ cwd: "/repo" });
+
+        expect(first.pr).toBeNull();
+        expect(second.pr?.number).toBe(42);
+        expect(state.statusCalls).toBe(2);
+        expect(state.detailsCalls).toBe(1);
+      }),
+    );
+  });
+
+  it("does not extend the remote metadata TTL when reusing cached remote status", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(0);
+    const state = {
+      currentDetails: baseDetails,
+      currentStatus: baseStatus,
+      detailsCalls: 0,
+      statusCalls: 0,
+    };
+
+    await runBroadcasterTest(
+      state,
+      Effect.gen(function* () {
+        const broadcaster = yield* GitStatusBroadcaster;
+
+        yield* broadcaster.getStatus({ cwd: "/repo" });
+        vi.setSystemTime(20_000);
+        yield* broadcaster.getStatus({ cwd: "/repo" });
+
+        vi.setSystemTime(31_000);
+        state.currentStatus = {
+          ...baseStatus,
+          pr: {
+            number: 43,
+            title: "Fresh PR",
+            url: "https://github.com/acme/repo/pull/43",
+            state: "open",
+          },
+        };
+        const third = yield* broadcaster.getStatus({ cwd: "/repo" });
+
+        expect(third.pr?.number).toBe(43);
+        expect(state.statusCalls).toBe(2);
+        expect(state.detailsCalls).toBe(2);
       }),
     );
   });
 
   it("refreshes the cached snapshot after explicit invalidation", async () => {
-    const state = { currentStatus: baseStatus, statusCalls: 0 };
+    const state = {
+      currentDetails: baseDetails,
+      currentStatus: baseStatus,
+      detailsCalls: 0,
+      statusCalls: 0,
+    };
 
     await runBroadcasterTest(
       state,
@@ -74,19 +208,30 @@ describe("GitStatusBroadcasterLive", () => {
           branch: "feature/updated-status",
           aheadCount: 2,
         };
+        state.currentDetails = {
+          ...baseDetails,
+          branch: "feature/updated-status",
+          aheadCount: 2,
+        };
         const refreshed = yield* broadcaster.refreshStatus("/repo");
         const cached = yield* broadcaster.getStatus({ cwd: "/repo" });
 
         expect(initial).toEqual(baseStatus);
         expect(refreshed).toEqual(state.currentStatus);
         expect(cached).toEqual(state.currentStatus);
         expect(state.statusCalls).toBe(2);
+        expect(state.detailsCalls).toBe(1);
       }),
     );
   });
 
   it("streams a status snapshot first and later refresh updates", async () => {
-    const state = { currentStatus: baseStatus, statusCalls: 0 };
+    const state = {
+      currentDetails: baseDetails,
+      currentStatus: baseStatus,
+      detailsCalls: 0,
+      statusCalls: 0,
+    };
 
     await runBroadcasterTest(
       state,