#1038 Add the read-only git adapter for worktree and base-branch ground truth

packages/fleet/README.md

@@ -4,6 +4,12 @@ Fleet is the server of the fleet-visibility stack. It watches the lifecycle-even
 
 A missing or empty events root serves an empty fleet. The folded state is a rebuildable in-server cache: restarting the server re-folds from disk, and nothing here is a writer concern.
 
+## Git adapter
+
+Alongside the event fold, a git adapter polls each lane's worktree (the `cwd` its events report) with read-only git commands and overlays the results on the lane: checked-out branch, working-tree change count, ahead/behind against the base branch (`behind > 0` is base-branch advance), and the base ref compared against. Observations are held in memory and merged at snapshot time — never written to the event store — and the adapter never fetches or otherwise mutates a repository, so base-branch advance is measured against the local remote-tracking ref, which worktrees share.
+
+A lane whose worktree no longer exists closes as `worktree-gone`. An unreadable repository degrades that lane's git fields to `null` while the rest of the fleet is unaffected.
+
 ## Run
 
 From `packages/fleet/`:
@@ -35,6 +41,7 @@ const client = hc<AppType>('http://localhost:4178');
 | Variable             | Default                  | Purpose                                                                                     |
 | -------------------- | ------------------------ | ------------------------------------------------------------------------------------------- |
 | `FLEET_EVENTS_DIR`   | `~/.codeassembly/events` | Root of the lifecycle-events tree to watch.                                                 |
+| `FLEET_GIT_POLL_MS`  | `15000`                  | Interval between read-only git polls of the lanes' worktrees.                               |
 | `FLEET_PORT`         | `4178`                   | Port to serve on.                                                                           |
 | `FLEET_RESCAN_MS`    | `5000`                   | Interval between full rescans — the correctness backstop when watching degrades.            |
 | `FLEET_RETENTION_MS` | `259200000`              | How long an idle lane is retained (≈ 3 days) before it is evicted and drops from the fleet. |

packages/fleet/src/__tests__/config.test.ts

@@ -11,6 +11,7 @@ describe('resolveConfig', () => {
       closeAfterMs: CLOSE_AFTER_MS,
       debounceMs: DEBOUNCE_MS,
       eventsDir: join(homedir(), '.codeassembly', 'events'),
+      gitPollMs: 15_000,
       heartbeatMs: HEARTBEAT_MS,
       port: 4178,
       rescanMs: 5000,
@@ -22,13 +23,15 @@ describe('resolveConfig', () => {
   it('reads every FLEET_* override from the environment', () => {
     const config = resolveConfig({
       FLEET_EVENTS_DIR: '/srv/events',
+      FLEET_GIT_POLL_MS: '2000',
       FLEET_PORT: '9000',
       FLEET_RESCAN_MS: '250',
       FLEET_RETENTION_MS: '600000',
       FLEET_STALE_MS: '1000',
     });
 
     expect(config.eventsDir).toBe('/srv/events');
+    expect(config.gitPollMs).toBe(2000);
     expect(config.port).toBe(9000);
     expect(config.rescanMs).toBe(250);
     expect(config.retentionMs).toBe(600_000);

packages/fleet/src/__tests__/e2e.test.ts

@@ -1,4 +1,5 @@
-import { appendFileSync, mkdirSync, mkdtempSync, rmSync } from 'node:fs';
+import { execFileSync } from 'node:child_process';
+import { appendFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 
@@ -11,22 +12,75 @@ import { type RunningFleetServer, startFleetServer } from '../server.ts';
 const SHORT_INTERVALS = {
   closeAfterMs: 600_000,
   debounceMs: 10,
+  gitPollMs: 60_000,
   heartbeatMs: 60_000,
   port: 0,
   rescanMs: 50,
   retentionMs: RETENTION_MS,
 };
 
 let eventsDir: string;
+let repoDir: string | undefined;
 let running: RunningFleetServer | undefined;
 
 /** Serializes one event envelope as a JSONL line. */
-function composeLine(type: string, ts: string): string {
-  return `${JSON.stringify({ id: '01ARZ3NDEKTSV4RRFFQ69G5FAV', ts, type, cwd: '/work/repo', payload: {} })}\n`;
+function composeLine(type: string, ts: string, cwd = '/work/repo'): string {
+  return `${JSON.stringify({ id: '01ARZ3NDEKTSV4RRFFQ69G5FAV', ts, type, cwd, payload: {} })}\n`;
+}
+
+/** Initializes a repository with one commit on `main` and its remote-tracking base ref in a fresh temp directory. */
+function createRepo(): string {
+  const dir = mkdtempSync(join(tmpdir(), 'fleet-e2e-repo-'));
+  execFileSync('git', ['-C', dir, 'init', '--initial-branch=main'], { stdio: 'ignore' });
+  execFileSync(
+    'git',
+    [
+      '-C',
+      dir,
+      '-c',
+      'user.name=fleet-test',
+      '-c',
+      'user.email=fleet@test.invalid',
+      '-c',
+      'commit.gpgsign=false',
+      'commit',
+      '--allow-empty',
+      '--message',
+      'one',
+    ],
+    { stdio: 'ignore' },
+  );
+  execFileSync('git', ['-C', dir, 'update-ref', 'refs/remotes/origin/main', 'HEAD'], { stdio: 'ignore' });
+  return dir;
+}
+
+/** Polls `/api/lanes` until the predicate holds on the first lane, failing the test after the timeout. */
+async function fetchLaneUntil(
+  port: number,
+  predicate: (lane: FleetSnapshot['lanes'][number]) => boolean,
+  timeoutMs = 3000,
+): Promise<FleetSnapshot['lanes'][number]> {
+  const deadline = Date.now() + timeoutMs;
+  let lane: FleetSnapshot['lanes'][number] | undefined;
+  while (Date.now() < deadline) {
+    const response = await fetch(`http://localhost:${port}/api/lanes`);
+    // eslint-disable-next-line @typescript-eslint/consistent-type-assertions -- shape is produced by the server under test; test-code carve-out
+    const snapshot = (await response.json()) as FleetSnapshot;
+    lane = snapshot.lanes[0];
+    if (lane !== undefined && predicate(lane)) {
+      return lane;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  assert(lane !== undefined, 'A lane should have appeared before the timeout');
+  expect(predicate(lane)).toBe(true);
+  return lane;
 }
 
 /** Starts a server on an ephemeral port over the given events root. */
-async function startTestServer(overrides: { eventsDir?: string; staleMs?: number } = {}): Promise<void> {
+async function startTestServer(
+  overrides: { eventsDir?: string; gitPollMs?: number; staleMs?: number } = {},
+): Promise<void> {
   running = await startFleetServer({
     config: { ...SHORT_INTERVALS, eventsDir, staleMs: 90_000, ...overrides },
     log: () => {},
@@ -68,6 +122,10 @@ afterEach(async () => {
   await running?.stop();
   running = undefined;
   rmSync(eventsDir, { recursive: true, force: true });
+  if (repoDir !== undefined) {
+    rmSync(repoDir, { recursive: true, force: true });
+    repoDir = undefined;
+  }
 });
 
 describe('fleet server', () => {
@@ -106,6 +164,25 @@ describe('fleet server', () => {
     expect(pushed.lanes[0]?.sessions[0]?.phase).toBe('working');
   });
 
+  it('surfaces git ground truth within a poll interval and closes the lane when the worktree disappears', async () => {
+    repoDir = createRepo();
+    writeFileSync(join(repoDir, 'wip.txt'), 'wip');
+    const laneDir = join(eventsDir, 'acme', 'app', '101');
+    mkdirSync(laneDir, { recursive: true });
+    appendFileSync(join(laneDir, 'sess-a.jsonl'), composeLine('turn.started', new Date().toISOString(), repoDir));
+    await startTestServer({ gitPollMs: 50 });
+    assert(running !== undefined, 'The server should be running');
+
+    const probed = await fetchLaneUntil(running.port, (lane) => lane.git !== null);
+    expect(probed.git).toEqual({ branch: 'main', dirtyFiles: 1, ahead: 0, behind: 0, baseBranch: 'origin/main' });
+    expect(probed.open).toBe(true);
+
+    rmSync(repoDir, { recursive: true, force: true });
+
+    const closed = await fetchLaneUntil(running.port, (lane) => !lane.open);
+    expect(closed.closedReason).toBe('worktree-gone');
+  });
+
   it('broadcasts a staleness crossing with no new event on disk', async () => {
     const laneDir = join(eventsDir, 'acme', 'app', '101');
     mkdirSync(laneDir, { recursive: true });