# Phase 2b: Services & Persistence (Tasks 13-18)

> **Prerequisites:** Phase 1 Tasks 1-5 complete. Read [conventions.md](conventions.md).
> **Parallel with:** Phase 2a (Session Stack).
> **Merge milestone:** M1 — after both Phase 2a and 2b complete, safe to merge (purely additive).

**Goal:** Convert InstanceManager to per-instance fibers, dissolve 5 leaf service classes into Effect Layers, switch persistence to @effect/sql-sqlite-node, complete OrchestrationEngine migration, add PTY stream, and add push notification service.

---

## Task 13: InstanceManager — per-instance fibers + acquireRelease

**Files:**
- Create: `src/lib/effect/instance-manager-service.ts`
- Test: `test/unit/instance/instance-manager-effect.test.ts`

**Step 1: Write the failing test**

```typescript
// test/unit/instance/instance-manager-effect.test.ts
import { describe, it } from "@effect/vitest";
import { expect, vi } from "vitest";
import { Effect, Layer, Ref, Fiber, Duration, Exit, HashMap } from "effect";
import {
  InstanceManagerStateTag,
  makeInstanceManagerStateLive,
  addInstance,
  removeInstance,
  type InstanceConfig,
} from "../../../src/lib/effect/instance-manager-service.js";

describe("InstanceManager Effect", () => {
  // AUDIT FIX (C-R5-3): healthPollers was moved to PollerFibersTag (FiberMap)
  // per C-NEW-4. Tests check instance state and FiberMap separately.
  it.scoped("addInstance registers instance in state", () =>
    Effect.gen(function* () {
      const config: InstanceConfig = {
        id: "inst-1",
        name: "Test Instance",
        port: 4096,
        managed: false,
      };

      yield* addInstance(config);
      const ref = yield* InstanceManagerStateTag;
      const state = yield* Ref.get(ref);
      expect(HashMap.has(state.instances, "inst-1")).toBe(true);
    }).pipe(Effect.provide(makeInstanceManagerStateLive()))
  );

  it.scoped("removeInstance clears instance from state", () =>
    Effect.gen(function* () {
      const config: InstanceConfig = {
        id: "inst-1",
        name: "Test Instance",
        port: 4096,
        managed: false,
      };

      yield* addInstance(config);
      yield* removeInstance("inst-1");
      const ref = yield* InstanceManagerStateTag;
      const state = yield* Ref.get(ref);
      expect(HashMap.has(state.instances, "inst-1")).toBe(false);
    }).pipe(Effect.provide(makeInstanceManagerStateLive()))
  );

  it.scoped("enforces max instance limit", () =>
    Effect.gen(function* () {
      const exit = yield* Effect.exit(
        Effect.gen(function* () {
          for (let i = 0; i < 6; i++) { // Max is 5
            yield* addInstance({
              id: `inst-${i}`, name: `Inst ${i}`, port: 4096 + i, managed: false,
            });
          }
        })
      );

      expect(Exit.isFailure(exit)).toBe(true);
    }).pipe(Effect.provide(makeInstanceManagerStateLive()))
  );
});
```

**Step 2: Run test to verify it fails**

Run: `pnpm vitest run test/unit/instance/instance-manager-effect.test.ts`
Expected: FAIL — module not found

**Step 3: Write implementation**

```typescript
// src/lib/effect/instance-manager-service.ts
// NOTE: Uses HashMap (not Map) per conventions — structural sharing in Ref.update.
import { Context, Data, Effect, Layer, Ref, Fiber, FiberMap, Schedule, Duration, HashMap } from "effect";
import { Command, CommandExecutor } from "@effect/platform";

// Import existing types from shared-types.ts. Do NOT redefine.
// InstanceConfig: { name, port, managed, env?, url? }
// OpenCodeInstance: InstanceConfig + { id, status, pid?, restartCount, createdAt, ... }
import type { InstanceConfig, OpenCodeInstance } from "../shared-types.js";

// AddInstanceInput includes the ID (generated by caller) plus InstanceConfig fields
export interface AddInstanceInput extends InstanceConfig {
  id: string;
}

export interface InstanceManagerConfig {
  maxInstances: number; // configurable, default 5
}

// AUDIT FIX (H8): Use FiberMap for healthPollers — auto-interrupts all
// fibers on scope close, provides run() for fork-and-register.
export interface InstanceManagerState {
  instances: HashMap.HashMap<string, OpenCodeInstance>;
  config: InstanceManagerConfig;
}

export class PollerFibersTag extends Context.Tag("PollerFibers")<
  PollerFibersTag,
  FiberMap.FiberMap<string>
>() {}

export class InstanceManagerStateTag extends Context.Tag("InstanceManagerState")<
  InstanceManagerStateTag,
  Ref.Ref<InstanceManagerState>
>() {}

export const makeInstanceManagerStateLive = (
  config: InstanceManagerConfig = { maxInstances: 5 }
): Layer.Layer<InstanceManagerStateTag | PollerFibersTag> =>
  Layer.mergeAll(
    Layer.effect(InstanceManagerStateTag, Ref.make({
      instances: HashMap.empty<string, OpenCodeInstance>(),
      config,
    })),
    Layer.scoped(PollerFibersTag, FiberMap.make<string>()),
  );

// --- Error types (Schema.TaggedError for catchTag support) ---

export class InstanceLimitExceeded extends Data.TaggedError("InstanceLimitExceeded")<{
  max: number;
}> {}

export class InstanceNotFound extends Data.TaggedError("InstanceNotFound")<{
  id: string;
}> {}

// --- Health polling ---
// Uses @effect/platform HttpClient for testability — tests inject a mock
// HttpClient Layer instead of hitting real ports.

import { HttpClient, HttpClientRequest } from "@effect/platform";

const healthPollFiber = (instanceId: string) =>
  Effect.gen(function* () {
    const ref = yield* InstanceManagerStateTag;
    const client = yield* HttpClient.HttpClient;

    const checkHealth = Effect.gen(function* () {
      const state = yield* Ref.get(ref);
      const instance = HashMap.get(state.instances, instanceId);
      // If instance removed, interrupt self to stop polling
      if (instance._tag === "None") return yield* Effect.interrupt;

      // @effect/platform HttpClient — injectable for tests
      const healthy = yield* client.execute(
        HttpClientRequest.get(`http://localhost:${instance.value.port}/health`)
      ).pipe(
        Effect.map((res) => res.status === 200),
        Effect.orElseSucceed(() => false),
      );

      yield* Ref.update(ref, (s) => {
        const existing = HashMap.get(s.instances, instanceId);
        if (existing._tag === "None") return s;
        return {
          ...s,
          instances: HashMap.set(s.instances, instanceId, {
            ...existing.value,
            status: healthy ? "healthy" : "unhealthy",
            lastHealthCheck: Date.now(),
          }),
        };
      });
    });

    // Retry individual poll failures (network errors, timeouts) — don't
    // let a single failed health check kill the polling fiber. Only catch
    // expected errors; defects (bugs) propagate to the supervisor.
    yield* checkHealth.pipe(
      Effect.catchAllCause((cause) =>
        // Log and continue — Cause includes both errors and defects for diagnostics
        Effect.logWarning("Health poll error for " + instanceId, cause)
      ),
      Effect.repeat(Schedule.spaced(Duration.seconds(5)))
    );
  });

// --- Process spawning via @effect/platform CommandExecutor ---
// Uses CommandExecutor instead of raw child_process.spawn so that
// tests can inject a mock executor without vi.mock("node:child_process").

export const spawnInstance = (instanceId: string) =>
  Effect.gen(function* () {
    const ref = yield* InstanceManagerStateTag;
    const state = yield* Ref.get(ref);
    const instance = HashMap.get(state.instances, instanceId);
    if (instance._tag === "None") return yield* Effect.fail(new InstanceNotFound({ id: instanceId }));

    const inst = instance.value;

    // Build command via @effect/platform Command
    const cmd = Command.make("opencode", "serve", "--port", String(inst.port)).pipe(
      Command.env(inst.env ?? {}),
    );

    // Start the process — CommandExecutor returns a running Process handle
    const executor = yield* CommandExecutor.CommandExecutor;
    const proc = yield* executor.start(cmd);

    // Register cleanup: the Process handle supports kill() which sends
    // SIGTERM. Wrap in a finalizer with escalation to SIGKILL after 5s.
    yield* Effect.addFinalizer(() =>
      proc.kill().pipe(
        Effect.timeout(Duration.seconds(5)),
        // If still alive after 5s, force kill
        Effect.catchAll(() =>
          Effect.sync(() => { try { process.kill(proc.pid, "SIGKILL"); } catch {} })
        ),
      )
    );

    // Update state with process info (using HashMap)
    yield* Ref.update(ref, (s) => {
      const existing = HashMap.get(s.instances, instanceId);
      if (existing._tag === "None") return s;
      return {
        ...s,
        instances: HashMap.set(s.instances, instanceId, { ...existing.value, status: "starting", pid: proc.pid }),
      };
    });

    // Start health polling
    yield* Effect.forkScoped(healthPollFiber(instanceId));

    return proc;
  });

// --- Restart with rate limiting ---

const restartSchedule = Schedule.exponential(Duration.seconds(1)).pipe(
  Schedule.intersect(Schedule.recurs(5)),
  Schedule.upTo(Duration.minutes(2))
);

export const restartWithLimit = (instanceId: string) =>
  spawnInstance(instanceId).pipe(Effect.retry(restartSchedule));

// --- CRUD operations ---

export const addInstance = (input: AddInstanceInput) =>
  Effect.gen(function* () {
    const { id, ...config } = input;
    const ref = yield* InstanceManagerStateTag;

    // Check capacity FIRST via atomic Ref.modify, THEN fork the poller.
    // This prevents a race where two concurrent addInstance calls both
    // fork fibers before either checks the limit.
    const limitError = yield* Ref.modify(ref, (s) => {
      if (HashMap.size(s.instances) >= s.config.maxInstances) {
        return [new InstanceLimitExceeded({ max: s.config.maxInstances }), s];
      }
      // Reserve the slot atomically (insert instance without poller yet)
      const newInstance: OpenCodeInstance = {
        ...config,
        id,
        status: "stopped",
        pid: undefined,
        restartCount: 0,
        createdAt: Date.now(),
        lastHealthCheck: undefined,
        exitCode: undefined,
        needsRestart: false,
      } as OpenCodeInstance;
      return [null, { ...s, instances: HashMap.set(s.instances, id, newInstance) }];
    });

    if (limitError) {
      return yield* Effect.fail(limitError);
    }

    // Slot reserved — now fork the health poller via FiberMap
    const fiberMap = yield* PollerFibersTag;
    yield* FiberMap.run(fiberMap, id, healthPollFiber(id));
  });

export const removeInstance = (instanceId: string) =>
  Effect.gen(function* () {
    const ref = yield* InstanceManagerStateTag;
    yield* Ref.update(ref, (s) => ({
      ...s,
      instances: HashMap.remove(s.instances, instanceId),
    }));
    // FiberMap.remove interrupts the fiber and removes the entry
    const fiberMap = yield* PollerFibersTag;
    yield* FiberMap.remove(fiberMap, instanceId);
  });

export const getInstance = (instanceId: string) =>
  Effect.gen(function* () {
    const ref = yield* InstanceManagerStateTag;
    const state = yield* Ref.get(ref);
    const instance = HashMap.get(state.instances, instanceId);
    if (instance._tag === "None") return yield* Effect.fail(new InstanceNotFound({ id: instanceId }));
    return instance.value;
  });

export const getInstances = Effect.gen(function* () {
  const ref = yield* InstanceManagerStateTag;
  const state = yield* Ref.get(ref);
  return HashMap.toEntries(state.instances).map(([_, v]) => v);
});

export const getInstanceUrl = (instanceId: string) =>
  Effect.gen(function* () {
    const instance = yield* getInstance(instanceId);
    return instance.url ?? `http://localhost:${instance.port}`;
  });

export const updateInstance = (instanceId: string, updates: Partial<InstanceConfig>) =>
  Effect.gen(function* () {
    const ref = yield* InstanceManagerStateTag;
    yield* Ref.update(ref, (s) => {
      const existing = HashMap.get(s.instances, instanceId);
      if (existing._tag === "None") return s;
      return { ...s, instances: HashMap.set(s.instances, instanceId, { ...existing.value, ...updates }) };
    });
  });

export const startInstance = (instanceId: string) =>
  spawnInstance(instanceId);

// stopInstance: interrupt the health poller fiber. The process finalizer
// (set in spawnInstance via Effect.addFinalizer) handles SIGTERM → SIGKILL.
export const stopInstance = (instanceId: string) =>
  Effect.gen(function* () {
    const ref = yield* InstanceManagerStateTag;
    yield* Ref.update(ref, (s) => {
      const existing = HashMap.get(s.instances, instanceId);
      return {
        ...s,
        instances: existing._tag === "Some"
          ? HashMap.set(s.instances, instanceId, { ...existing.value, status: "stopped", pid: undefined })
          : s.instances,
      };
    });
    // FiberMap.remove interrupts the fiber and removes the entry
    const fiberMap = yield* PollerFibersTag;
    yield* FiberMap.remove(fiberMap, instanceId);
  });

// stopAll: FiberMap auto-interrupts all fibers on scope close.
// Explicit stopAll clears the map for mid-lifecycle shutdown.
export const stopAll = Effect.gen(function* () {
  const fiberMap = yield* PollerFibersTag;
  yield* FiberMap.clear(fiberMap);
});
```

**Step 4: Update tests to cover additional methods and edge cases**

Add to the test file:
```typescript
  it.scoped("add-remove-add at max instances succeeds", () =>
    Effect.gen(function* () {
      // Add 5 instances (max)
      for (let i = 0; i < 5; i++) {
        yield* addInstance({
          id: `inst-${i}`, name: `Inst ${i}`, port: 4096 + i, managed: false,
        });
      }
      // Remove one
      yield* removeInstance("inst-0");
      // Add one more — should succeed
      yield* addInstance({
        id: "inst-new", name: "New", port: 5000, managed: false,
      });
      const ref = yield* InstanceManagerStateTag;
      const state = yield* Ref.get(ref);
      expect(HashMap.size(state.instances)).toBe(5);
    }).pipe(Effect.provide(makeInstanceManagerStateLive()))
  );

  it.scoped("max instance error has correct tag for catchTag", () =>
    Effect.gen(function* () {
      const result = yield* Effect.gen(function* () {
        for (let i = 0; i < 6; i++) {
          yield* addInstance({
            id: `inst-${i}`, name: `Inst ${i}`, port: 4096 + i, managed: false,
          });
        }
      }).pipe(
        Effect.catchTag("InstanceLimitExceeded", (e) =>
          Effect.succeed(`caught: max=${e.max}`)
        )
      );

      expect(result).toBe("caught: max=5");
    }).pipe(Effect.provide(makeInstanceManagerStateLive()))
  );
```

**Step 5: Run test to verify it passes**

Run: `pnpm vitest run test/unit/instance/instance-manager-effect.test.ts`
Expected: 5 tests PASS

**Step 6: Commit**

```bash
git add src/lib/effect/instance-manager-service.ts test/unit/instance/instance-manager-effect.test.ts
git commit -m "feat(effect): replace InstanceManager with per-instance fibers + acquireRelease + SIGTERM/SIGKILL"
```

---

## Task 14: Leaf services — dissolve classes into Layers

> **NOTE:** `TrackedService` and `AsyncTracker` were already deleted on this
> branch (commit `a70d53b`). These leaf services are now plain classes with
> callback maps or drain methods. This task converts their CLASS INTERNALS
> to Effect primitives (Ref, Schedule, Fiber) and wraps them in Layers.
>
> **IMPORTANT: Execute as 5 separate sub-tasks (14a-14e), each with its own
> commit.** Each leaf service has unique internals (platform detection,
> semver comparison, hysteresis logic) that should be verified independently.
> Do NOT bundle all 5 into one commit — a failure in one should not block others.
>
> **Sub-tasks:**
> - **14a:** StorageMonitor — state-machine transitions (unknown->low, low->ok)
> - **14b:** VersionChecker — semver comparison, broadcast on change
> - **14c:** PortScanner — hysteresis (consecutive failure threshold)
> - **14d:** KeepAwake — platform detection, process group killing
> - **14e:** RateLimiter — sliding window token bucket
>
> Each sub-task follows the same pattern:
> 1. Write failing test for the new Layer
> 2. Write the Effect Layer implementation
> 3. Update the Tag in `services.ts` to match the new interface
> 4. Run `pnpm vitest run test/unit/daemon/<service>.test.ts`
> 5. Commit: `feat(effect): dissolve <ServiceName> class into Effect Layer`

> **Update Tag interfaces.** Each leaf service Tag in `services.ts`
> currently references the old imperative class type (e.g., `StorageMonitor`,
> `VersionChecker`). As part of each sub-task, update the Tag to use the new
> Effect-native service interface. Delete the old class file and update all
> import sites. No `as any` casts — the types must align cleanly.

**Files (across all sub-tasks):**
- Create: `src/lib/effect/storage-monitor-layer.ts`
- Create: `src/lib/effect/version-checker-layer.ts`
- Create: `src/lib/effect/port-scanner-layer.ts`
- Create: `src/lib/effect/keep-awake-layer.ts`
- Create: `src/lib/effect/rate-limiter-layer.ts`
- Modify: `src/lib/effect/services.ts` (update Tag payload types)
- Delete: `src/lib/daemon/storage-monitor.ts`, `version-check.ts`, `port-scanner.ts`, `keep-awake.ts`
- Test: `test/unit/daemon/leaf-services-effect.test.ts` (or split into per-service test files)

Showing StorageMonitor (14a) in detail; others repeat the pattern.

**Step 1: Write the failing test**

```typescript
// test/unit/daemon/leaf-services-effect.test.ts
import { describe, it } from "@effect/vitest";
import { expect, vi } from "vitest";
import { Effect, Layer, Exit, Duration, TestClock, Fiber } from "effect";
import { StorageMonitorTag, StorageMonitorLive } from "../../../src/lib/effect/storage-monitor-layer.js";
import { VersionCheckerTag, VersionCheckerLive } from "../../../src/lib/effect/version-checker-layer.js";
import { RateLimiterTag, RateLimiterLive } from "../../../src/lib/effect/rate-limiter-layer.js";

describe("Leaf service Layers", () => {
  describe("StorageMonitor", () => {
    it.scoped("Layer constructs and provides tag", () =>
      Effect.gen(function* () {
        const mockPersistence = {
          evictOldEvents: vi.fn().mockReturnValue(Effect.succeed(undefined)),
        };
        const mockGetUsage = vi.fn().mockReturnValue(Effect.succeed(0.5)); // 50% usage

        const exit = yield* Effect.exit(
          Effect.gen(function* () {
            // Layer should start without error
            yield* Effect.sleep(Duration.millis(10));
          })
        );

        expect(Exit.isSuccess(exit)).toBe(true);
      }).pipe(
        Effect.provide(StorageMonitorLive({
          getStorageUsage: vi.fn().mockReturnValue(Effect.succeed(0.5)),
          persistence: { evictOldEvents: vi.fn().mockReturnValue(Effect.succeed(undefined)) } as any,
          checkInterval: Duration.seconds(60),
          highWaterMark: 0.9,
        }))
      )
    );
  });

  describe("RateLimiter", () => {
    it.scoped("allows requests under limit", () =>
      Effect.gen(function* () {
        const limiter = yield* RateLimiterTag;
        const r1 = yield* limiter.checkLimit("127.0.0.1");
        const r2 = yield* limiter.checkLimit("127.0.0.1");

        expect(r1).toBe(true);
        expect(r2).toBe(true);
      }).pipe(Effect.provide(RateLimiterLive({ maxRequests: 5, windowMs: 10_000 })))
    );

    it.scoped("blocks requests over limit", () =>
      Effect.gen(function* () {
        const limiter = yield* RateLimiterTag;
        for (let i = 0; i < 5; i++) {
          yield* limiter.checkLimit("127.0.0.1");
        }
        const result = yield* limiter.checkLimit("127.0.0.1"); // 6th request

        expect(result).toBe(false);
      }).pipe(Effect.provide(RateLimiterLive({ maxRequests: 5, windowMs: 10_000 })))
    );
  });
});
```

**Step 2: Run test to verify it fails**

Run: `pnpm vitest run test/unit/daemon/leaf-services-effect.test.ts`
Expected: FAIL — modules not found

**Step 3: Write StorageMonitor Layer**

```typescript
// src/lib/effect/storage-monitor-layer.ts
// NOTE: Re-uses existing StorageMonitorTag from services.ts.
// The actual StorageMonitor class (136 lines) has state-machine transitions
// (unknown->low, low->ok) and only emits events on transitions.
// The StorageMonitorTag in services.ts MUST be updated to use this
// Effect-native interface instead of the old class type.
import { Context, Effect, Layer, Ref, Schedule, Duration } from "effect";

interface StorageMonitorConfig {
  getStorageUsage: () => Effect.Effect<number>;
  persistence: { evictOldEvents: () => Effect.Effect<void> };
  checkInterval: Duration.DurationInput;
  highWaterMark: number;
}

interface StorageMonitorService {
  getUsage: () => Effect.Effect<number>;
  getLastCheck: () => Effect.Effect<number>;
}

// Provides StorageMonitorTag so other services can query usage.
// NOT Layer.scopedDiscard — that would fail to provide the Tag.
export const StorageMonitorLive = (config: StorageMonitorConfig) =>
  Layer.scoped(
    StorageMonitorTag,
    Effect.gen(function* () {
      const state = yield* Ref.make({ lastCheck: 0, usage: 0 });

      const check = Effect.gen(function* () {
        const usage = yield* config.getStorageUsage();
        yield* Ref.set(state, { lastCheck: Date.now(), usage });
        if (usage > config.highWaterMark) {
          yield* config.persistence.evictOldEvents();
        }
      });

      // Background fiber — retries on unexpected errors
      yield* check.pipe(
        Effect.repeat(Schedule.spaced(config.checkInterval)),
        Effect.retry(Schedule.exponential("5 seconds").pipe(Schedule.intersect(Schedule.recurs(3)))),
        Effect.catchAll((e) => Effect.logWarning("Storage monitor failed after retries", e)),
        Effect.forkScoped
      );

      // Expose read-only access to state
      return {
        getUsage: () => Ref.get(state).pipe(Effect.map((s) => s.usage)),
        getLastCheck: () => Ref.get(state).pipe(Effect.map((s) => s.lastCheck)),
      }; // StorageMonitorTag interface updated to match this shape
    })
  );
```

**Step 4: Write RateLimiter Layer**

```typescript
// src/lib/effect/rate-limiter-layer.ts
// NOTE: Uses HashMap (not Map) per conventions — structural sharing in Ref.update.
import { Context, Effect, Layer, Ref, Schedule, Duration, HashMap, Chunk } from "effect";

interface RateLimiterService {
  checkLimit: (ip: string) => Effect.Effect<boolean>;
}

export class RateLimiterTag extends Context.Tag("RateLimiter")<
  RateLimiterTag,
  RateLimiterService
>() {}

interface BucketEntry {
  tokens: ReadonlyArray<number>;
}

interface RateLimiterState {
  buckets: HashMap.HashMap<string, BucketEntry>;
}

interface RateLimiterConfig {
  maxRequests: number;
  windowMs: number;
}

const tryConsume = (ip: string, config: RateLimiterConfig, now: number) =>
  (state: RateLimiterState): [boolean, RateLimiterState] => {
    const existing = HashMap.get(state.buckets, ip);
    const entry = existing._tag === "Some" ? existing.value : { tokens: [] as number[] };

    // Remove expired tokens
    const validTokens = entry.tokens.filter((t) => now - t < config.windowMs);

    if (validTokens.length >= config.maxRequests) {
      return [false, { buckets: HashMap.set(state.buckets, ip, { tokens: validTokens }) }];
    }

    return [true, { buckets: HashMap.set(state.buckets, ip, { tokens: [...validTokens, now] }) }];
  };

const evictStale = (windowMs: number, now: number) =>
  (state: RateLimiterState): RateLimiterState => {
    let buckets = HashMap.empty<string, BucketEntry>();
    for (const [ip, entry] of HashMap.toEntries(state.buckets)) {
      const valid = entry.tokens.filter((t) => now - t < windowMs);
      if (valid.length > 0) buckets = HashMap.set(buckets, ip, { tokens: valid });
    }
    return { buckets };
  };

export const RateLimiterLive = (config: RateLimiterConfig) =>
  Layer.scoped(
    RateLimiterTag,
    Effect.gen(function* () {
      const state = yield* Ref.make<RateLimiterState>({ buckets: HashMap.empty() });

      // Cleanup stale entries every 60s
      yield* Effect.sync(() => Date.now()).pipe(
        Effect.flatMap((now) => Ref.update(state, evictStale(config.windowMs, now))),
        Effect.repeat(Schedule.spaced(Duration.minutes(1))),
        Effect.forkScoped
      );

      return {
        checkLimit: (ip: string) =>
          Effect.sync(() => Date.now()).pipe(
            Effect.flatMap((now) => Ref.modify(state, tryConsume(ip, config, now)))
          ),
      };
    })
  );
```

**Step 5: Write VersionChecker, PortScanner, KeepAwake Layers** (same pattern as StorageMonitor)

```typescript
// src/lib/effect/version-checker-layer.ts
// NOTE: Re-uses existing VersionCheckerTag from services.ts.
// The actual VersionChecker class (298 lines) includes semver comparison
// logic (isNewer, parseSemver, comparePrereleases). When dissolving the
// class, either extract the semver functions as pure utilities or use
// an npm semver package.
import { Context, Effect, Layer, Ref, Schedule, Duration } from "effect";
import { VersionCheckerTag } from "./services.js";

interface VersionCheckerConfig {
  getCurrentVersion: () => string;
  fetchLatestVersion: () => Effect.Effect<string | null>;
  broadcast: (msg: { type: string; current: string; latest: string }) => Effect.Effect<void>;
  checkInterval: Duration.DurationInput;
}

// Semver comparison — extract from existing version-check.ts:parseSemver/isNewer
const isNewerVersion = (current: string, latest: string): boolean => {
  // Strip leading 'v'
  const c = current.replace(/^v/, "").split(".");
  const l = latest.replace(/^v/, "").split(".");
  for (let i = 0; i < 3; i++) {
    const cv = parseInt(c[i] ?? "0", 10);
    const lv = parseInt(l[i] ?? "0", 10);
    if (lv > cv) return true;
    if (lv < cv) return false;
  }
  return false; // Equal
};

// Provides VersionCheckerTag so daemon status can report latest version.
export const VersionCheckerLive = (config: VersionCheckerConfig) =>
  Layer.scoped(
    VersionCheckerTag,
    Effect.gen(function* () {
      const latestKnown = yield* Ref.make<string | null>(null);

      const check = Effect.gen(function* () {
        const latest = yield* config.fetchLatestVersion();
        if (latest) {
          const prev = yield* Ref.get(latestKnown);
          const current = config.getCurrentVersion();
          if (latest !== prev && isNewerVersion(current, latest)) {
            yield* Ref.set(latestKnown, latest);
            yield* config.broadcast({ type: "version_update", current, latest });
          }
        }
      });

      yield* check.pipe(
        Effect.repeat(Schedule.spaced(config.checkInterval)),
        Effect.retry(Schedule.exponential("10 seconds").pipe(Schedule.intersect(Schedule.recurs(3)))),
        Effect.catchAll((e) => Effect.logWarning("Version checker failed after retries", e)),
        Effect.forkScoped
      );

      return {
        getLatestKnown: () => Ref.get(latestKnown),
        getCurrentVersion: () => Effect.succeed(config.getCurrentVersion()),
      }; // VersionCheckerTag interface updated to match this shape
    })
  );
```

```typescript
// src/lib/effect/port-scanner-layer.ts
// NOTE: Re-uses existing PortScannerTag from services.ts.
// The actual PortScanner class (116 lines) has hysteresis: a port must
// fail `removalThreshold` consecutive times before being removed. This
// prevents flapping when a service briefly restarts.
import { Context, Effect, Layer, Ref, Schedule, Duration } from "effect";
import { PortScannerTag } from "./services.js";

interface PortScannerConfig {
  probeFn: (port: number) => Effect.Effect<boolean>;
  portRange: [number, number];
  scanInterval: Duration.DurationInput;
  removalThreshold: number; // Consecutive failures before removal
  onDiscovered: (port: number) => Effect.Effect<void>;
  onLost: (port: number) => Effect.Effect<void>;
  excludedPorts?: Set<number>;
}

// AUDIT FIX (L5): Uses native Set/Map (not HashMap/HashSet) because:
// 1. Set<number> iteration order matters for deterministic scan sequences
// 2. Map<number, number> eviction logic relies on insertion-order iteration
// 3. Values are plain numbers, not mutable Effect primitives
// Documented exception per conventions.
interface ScanState {
  knownPorts: Set<number>;
  failureCounts: Map<number, number>; // port -> consecutive failure count
}

// Provides PortScannerTag so daemon can query discovered ports.
export const PortScannerLive = (config: PortScannerConfig) =>
  Layer.scoped(
    PortScannerTag,
    Effect.gen(function* () {
      const state = yield* Ref.make<ScanState>({
        knownPorts: new Set(),
        failureCounts: new Map(),
      });

      const runScan = Effect.gen(function* () {
        const [start, end] = config.portRange;
        const excluded = config.excludedPorts ?? new Set();
        const ports = Array.from(
          { length: end - start + 1 },
          (_, i) => start + i
        ).filter((p) => !excluded.has(p));

        // Probe all ports concurrently
        const results = yield* Effect.forEach(
          ports,
          (port) => config.probeFn(port).pipe(
            Effect.map((alive) => ({ port, alive })),
            Effect.catchAll(() => Effect.succeed({ port, alive: false }))
          ),
          { concurrency: 10 }
        );

        const current = yield* Ref.get(state);
        const newKnown = new Set(current.knownPorts);
        const newFailures = new Map(current.failureCounts);

        for (const { port, alive } of results) {
          if (alive) {
            newFailures.delete(port);
            if (!current.knownPorts.has(port)) {
              newKnown.add(port);
              yield* config.onDiscovered(port).pipe(
                Effect.catchAll((e) => Effect.logWarning("onDiscovered error", e))
              );
            }
          } else if (current.knownPorts.has(port)) {
            // Hysteresis: increment failure count, only remove after threshold
            const failures = (newFailures.get(port) ?? 0) + 1;
            if (failures >= config.removalThreshold) {
              newKnown.delete(port);
              newFailures.delete(port);
              yield* config.onLost(port).pipe(
                Effect.catchAll((e) => Effect.logWarning("onLost error", e))
              );
            } else {
              newFailures.set(port, failures);
            }
          }
        }

        yield* Ref.set(state, { knownPorts: newKnown, failureCounts: newFailures });
      });

      yield* runScan.pipe(
        Effect.repeat(Schedule.spaced(config.scanInterval)),
        Effect.retry(Schedule.exponential("5 seconds").pipe(Schedule.intersect(Schedule.recurs(3)))),
        Effect.catchAll((e) => Effect.logWarning("Port scanner failed after retries", e)),
        Effect.forkScoped
      );

      return {
        getKnownPorts: () => Ref.get(state).pipe(Effect.map((s) => s.knownPorts)),
      }; // PortScannerTag interface updated to match this shape
    })
  );
```

```typescript
// src/lib/effect/keep-awake-layer.ts
// NOTE: The existing KeepAwake class (src/lib/daemon/keep-awake.ts, 249 lines)
// already has platform detection, process group killing, command resolution
// caching, and error events. The daemon-layers.ts Layer wraps it with drain
// finalizer. This task converts the CLASS INTERNALS to Effect primitives.
// The KeepAwakeTag already exists in services.ts and daemon-layers.ts.
//
// For now, keep using the existing KeepAwake class wrapped in the Layer
// (as done in daemon-layers.ts:makeKeepAwakeLive). A future task can
// dissolve the class internals into Effect.Schedule + Ref + Command.
//
// When dissolving, the key logic to preserve:
// - Platform detection: macOS -> "caffeinate -di", Linux -> "systemd-inhibit"
// - Command resolution caching (resolveCommand is called once, cached)
// - Process group killing: process.kill(-pid, "SIGTERM") not proc.kill()
// - Error event on unexpected exit
// - Idempotent activate/deactivate

import { Context, Effect, Fiber, Layer, Ref } from "effect";
import { Command, CommandExecutor } from "@effect/platform";

interface KeepAwakeService {
  activate: () => Effect.Effect<void>;
  deactivate: () => Effect.Effect<void>;
  isActive: () => Effect.Effect<boolean>;
  isSupported: () => Effect.Effect<boolean>;
}

// Re-use existing KeepAwakeTag from services.ts
import { KeepAwakeTag } from "./services.js";

interface KeepAwakeConfig {
  command?: string;
  args?: string[];
}

const detectPlatformCommand = (): { command: string; args: string[] } | null => {
  switch (process.platform) {
    case "darwin": return { command: "caffeinate", args: ["-di"] };
    case "linux": return {
      command: "systemd-inhibit",
      args: ["--what=idle", "--who=conduit", "--why=Keeping system awake", "sleep", "infinity"],
    };
    default: return null;
  }
};

export const KeepAwakeLive = (config?: KeepAwakeConfig) =>
  Layer.scoped(
    KeepAwakeTag,
    Effect.gen(function* () {
      const fiberRef = yield* Ref.make<Fiber.RuntimeFiber<void> | null>(null);
      const activeRef = yield* Ref.make(false);

      // Resolve command: user config or platform detection (cached)
      const resolved = config?.command
        ? { command: config.command, args: config.args ?? [] }
        : detectPlatformCommand();

      const activate = Effect.gen(function* () {
        if (!resolved) return; // Unsupported platform
        const alreadyActive = yield* Ref.get(activeRef);
        if (alreadyActive) return; // Idempotent

        // AUDIT FIX (L-R5-3): Use CommandExecutor.start (not Command.start)
        // for consistency with Task 13's pattern. Command.make builds the
        // command spec; CommandExecutor.start runs it.
        const cmd = Command.make(resolved.command, ...resolved.args);
        const executor = yield* CommandExecutor.CommandExecutor;
        const fiber = yield* executor.start(cmd).pipe(
          Effect.flatMap((proc) => proc.exitCode),
          Effect.catchAll((e) => Effect.logWarning("Keep-awake process exited", e)),
          Effect.forkScoped
        );
        yield* Ref.set(fiberRef, fiber);
        yield* Ref.set(activeRef, true);
      });

      const deactivate = Effect.gen(function* () {
        const fiber = yield* Ref.get(fiberRef);
        if (fiber) yield* Fiber.interrupt(fiber);
        yield* Ref.set(fiberRef, null);
        yield* Ref.set(activeRef, false);
      });

      // Finalizer: deactivate on scope close
      yield* Effect.addFinalizer(() => deactivate);

      return {
        activate,
        deactivate,
        isActive: () => Ref.get(activeRef),
        isSupported: () => Effect.succeed(resolved !== null),
      }; // KeepAwakeTag interface updated to match this shape
    })
  );
```

**Step 6: Run tests**

Run: `pnpm vitest run test/unit/daemon/leaf-services-effect.test.ts`
Expected: All tests PASS

**Step 7: Commit — AUDIT FIX (H-R5-6): One commit per sub-task**

> **IMPORTANT:** Despite the combined step 7 below, each sub-task (14a-14e)
> MUST be committed separately as stated in the task header. The combined
> git add below is shown for reference only. In practice:
> - 14a commit: `feat(effect): dissolve StorageMonitor class into Effect Layer`
> - 14b commit: `feat(effect): dissolve VersionChecker class into Effect Layer`
> - 14c commit: `feat(effect): dissolve PortScanner class into Effect Layer`
> - 14d commit: `feat(effect): dissolve KeepAwake class into Effect Layer`
> - 14e commit: `feat(effect): dissolve RateLimiter class into Effect Layer`

```bash
# Example for 14a only — repeat pattern for 14b-14e:
git add src/lib/effect/storage-monitor-layer.ts test/unit/daemon/leaf-services-effect.test.ts
git commit -m "feat(effect): dissolve StorageMonitor class into Effect Layer"
```

---

> **AUDIT FIX:** Create a checkpoint branch before Task 15:
> ```bash
> git branch checkpoint/pre-persistence-migration
> ```
> If the persistence migration fails, restore: `git checkout checkpoint/pre-persistence-migration -- .`

## Task 15: Persistence Layer — @effect/sql-sqlite-node

> **Step 0:** Install: `pnpm add @effect/sql@0.30.0 @effect/sql-sqlite-node@0.28.0 better-sqlite3 && pnpm add -D @types/better-sqlite3`
> The codebase currently uses Node 22+ native `node:sqlite` (`DatabaseSync`).
> This task switches to `better-sqlite3` via `@effect/sql-sqlite-node` — no
> wrappers needed. `SqlClient` provides type-safe tagged template queries,
> built-in `SqlClient.withTransaction`, and connection pooling as first-class
> Effect primitives. All existing `DatabaseSync` call sites migrate to
> `yield* sql\`...\`` — the old `node:sqlite` wrapper is deleted in Task 20m.

**Files:**
- Create: `src/lib/effect/persistence-service.ts`
- Test: `test/unit/persistence/persistence-effect.test.ts`

**Step 1: Write the failing test**

```typescript
// test/unit/persistence/persistence-effect.test.ts
import { describe, it } from "@effect/vitest";
import { expect } from "vitest";
import { Effect, Layer } from "effect";
import { SqlClient } from "@effect/sql";
import { SqliteClient } from "@effect/sql-sqlite-node";
import {
  PersistenceServiceTag,
  makePersistenceServiceLive,
  withTransaction,
} from "../../../src/lib/effect/persistence-service.js";

const TestSqlLayer = SqliteClient.layer({ filename: ":memory:" });

describe("Persistence Effect", () => {
  it.effect("migrate creates events table and indexes", () =>
    Effect.gen(function* () {
      const persistence = yield* PersistenceServiceTag;
      yield* persistence.migrate;
      const sql = yield* SqlClient.SqlClient;
      // Table should exist — insert a row to verify
      yield* sql`INSERT INTO events (type, payload) VALUES ('test', '{}')`;
      const rows = yield* sql`SELECT * FROM events`;
      expect(rows.length).toBe(1);
    }).pipe(Effect.provide(Layer.provideMerge(TestSqlLayer, makePersistenceServiceLive)))
  );

  it.effect("healthCheck returns true when DB is accessible", () =>
    Effect.gen(function* () {
      const persistence = yield* PersistenceServiceTag;
      const healthy = yield* persistence.healthCheck;
      expect(healthy).toBe(true);
    }).pipe(Effect.provide(Layer.provideMerge(TestSqlLayer, makePersistenceServiceLive)))
  );

  it.effect("evictBefore deletes old events", () =>
    Effect.gen(function* () {
      const persistence = yield* PersistenceServiceTag;
      yield* persistence.migrate;
      const sql = yield* SqlClient.SqlClient;
      // Insert events with known timestamps
      yield* sql`INSERT INTO events (type, payload, created_at) VALUES ('old', '{}', 1000)`;
      yield* sql`INSERT INTO events (type, payload, created_at) VALUES ('new', '{}', 9999999999)`;
      const deleted = yield* persistence.evictBefore(5000);
      expect(deleted).toBe(1);
      const remaining = yield* sql`SELECT * FROM events`;
      expect(remaining.length).toBe(1);
    }).pipe(Effect.provide(Layer.provideMerge(TestSqlLayer, makePersistenceServiceLive)))
  );

  it.effect("withTransaction commits on success", () =>
    Effect.gen(function* () {
      const sql = yield* SqlClient.SqlClient;
      yield* sql`CREATE TABLE test_items (id INTEGER PRIMARY KEY, name TEXT)`;
      yield* withTransaction(sql`INSERT INTO test_items (id, name) VALUES (1, 'item-1')`);
      const rows = yield* sql`SELECT * FROM test_items`;
      expect(rows.length).toBe(1);
    }).pipe(Effect.provide(TestSqlLayer))
  );

  it.effect("withTransaction rolls back on failure", () =>
    Effect.gen(function* () {
      const sql = yield* SqlClient.SqlClient;
      yield* sql`CREATE TABLE test_rollback (id INTEGER PRIMARY KEY, name TEXT)`;
      yield* withTransaction(
        Effect.gen(function* () {
          yield* sql`INSERT INTO test_rollback (id, name) VALUES (1, 'x')`;
          yield* Effect.fail(new Error("boom"));
        })
      ).pipe(Effect.catchAll(() => Effect.void));
      const rows = yield* sql`SELECT * FROM test_rollback`;
      expect(rows.length).toBe(0);
    }).pipe(Effect.provide(TestSqlLayer))
  );
});
```

**Step 2: Run test to verify it fails**

Run: `pnpm vitest run test/unit/persistence/persistence-effect.test.ts`
Expected: FAIL — module not found

**Step 3: Write implementation**

> **AUDIT FIX (M7):** The original PersistenceService was a trivial wrapper around
> SqlClient that added no value. This version adds real functionality: schema
> migration management, typed query helpers for the event store, and a health
> check. If the service doesn't add value beyond SqlClient, don't create the
> indirection — use SqlClient.SqlClient directly.

```typescript
// src/lib/effect/persistence-service.ts
import { Context, Data, Effect, Layer } from "effect";
import { SqlClient } from "@effect/sql";

export class PersistenceError extends Data.TaggedError("PersistenceError")<{
  operation: string;
  cause: unknown;
}> {}

export interface PersistenceService {
  /** Run the event store schema migration (CREATE TABLE IF NOT EXISTS). */
  readonly migrate: Effect.Effect<void, PersistenceError>;
  /** Check SQLite is accessible and writable. */
  readonly healthCheck: Effect.Effect<boolean, PersistenceError>;
  /** Evict events older than the given timestamp. Returns count deleted. */
  readonly evictBefore: (timestampMs: number) => Effect.Effect<number, PersistenceError>;
  /** Access the raw SqlClient for ad-hoc queries in projectors. */
  readonly sql: SqlClient.SqlClient;
}

export class PersistenceServiceTag extends Context.Tag("PersistenceService")<
  PersistenceServiceTag,
  PersistenceService
>() {}

export const makePersistenceServiceLive: Layer.Layer<PersistenceServiceTag, never, SqlClient.SqlClient> =
  Layer.effect(
    PersistenceServiceTag,
    Effect.gen(function* () {
      const sql = yield* SqlClient.SqlClient;

      const migrate = Effect.gen(function* () {
        yield* sql`CREATE TABLE IF NOT EXISTS events (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          type TEXT NOT NULL,
          session_id TEXT,
          payload TEXT NOT NULL,
          created_at INTEGER NOT NULL DEFAULT (unixepoch('now','subsec') * 1000)
        )`;
        yield* sql`CREATE INDEX IF NOT EXISTS idx_events_session ON events(session_id)`;
        yield* sql`CREATE INDEX IF NOT EXISTS idx_events_created ON events(created_at)`;
      }).pipe(
        Effect.mapError((e) => new PersistenceError({ operation: "migrate", cause: e })),
        Effect.withSpan("persistence.migrate")
      );

      const healthCheck = sql`SELECT 1 AS ok`.pipe(
        Effect.map((rows) => rows.length > 0),
        Effect.mapError((e) => new PersistenceError({ operation: "healthCheck", cause: e })),
        Effect.withSpan("persistence.healthCheck")
      );

      const evictBefore = (timestampMs: number) =>
        sql`DELETE FROM events WHERE created_at < ${timestampMs}`.pipe(
          Effect.map((result) => (result as any).changes ?? 0),
          Effect.mapError((e) => new PersistenceError({ operation: "evictBefore", cause: e })),
          Effect.withSpan("persistence.evictBefore")
        );

      return { migrate, healthCheck, evictBefore, sql };
    })
  );

// AUDIT FIX (M-R5-4): SqlClient.withTransaction is NOT a static export.
// Transactions are run via the SqlClient instance: `yield* sql.withTransaction(effect)`.
// Export a helper that accesses the SqlClient from context:
export const withTransaction = <A, E, R>(
  effect: Effect.Effect<A, E, R>
): Effect.Effect<A, E | PersistenceError, R | SqlClient.SqlClient> =>
  Effect.gen(function* () {
    const sql = yield* SqlClient.SqlClient;
    return yield* sql.withTransaction(effect);
  }).pipe(Effect.mapError((e) =>
    e instanceof PersistenceError ? e : new PersistenceError({ operation: "transaction", cause: e })
  ));
```

**Step 4: Run test to verify it passes**

Run: `pnpm vitest run test/unit/persistence/persistence-effect.test.ts`
Expected: 2 tests PASS

**Step 5: Commit**

```bash
git add src/lib/effect/persistence-service.ts test/unit/persistence/persistence-effect.test.ts
git commit -m "feat(effect): add @effect/sql-sqlite-node persistence with built-in transactions"
```

---

## Task 16: OrchestrationEngine — complete partial migration

**Files:**
- Create: `src/lib/effect/orchestration-service.ts`
- Test: `test/unit/provider/orchestration-effect.test.ts`

**Step 1: Write the failing test**

```typescript
// test/unit/provider/orchestration-effect.test.ts
import { describe, it } from "@effect/vitest";
import { expect, vi } from "vitest";
import { Effect, Layer, Ref } from "effect";
import {
  IdempotencySetTag,
  makeIdempotencySetLive,
  routeCommand,
} from "../../../src/lib/effect/orchestration-service.js";

describe("OrchestrationEngine Effect", () => {
  it.effect("routes command to provider", () =>
    Effect.gen(function* () {
      const mockProvider = {
        execute: vi.fn().mockReturnValue(
          Effect.succeed({ text: "response" })
        ),
      };

      const result = yield* routeCommand(
        { id: "cmd-1", type: "send_turn", payload: "hello" },
        mockProvider as any
      );

      expect(result).toEqual({ text: "response" });
      expect(mockProvider.execute).toHaveBeenCalled();
    }).pipe(Effect.provide(makeIdempotencySetLive()))
  );

  it.effect("deduplicates repeated command IDs", () =>
    Effect.gen(function* () {
      const mockProvider = {
        execute: vi.fn().mockReturnValue(Effect.succeed({ text: "response" })),
      };

      yield* routeCommand({ id: "cmd-1", type: "send_turn", payload: "hello" }, mockProvider as any);
      const result = yield* routeCommand({ id: "cmd-1", type: "send_turn", payload: "hello" }, mockProvider as any);

      expect(result).toEqual({ deduplicated: true });
      expect(mockProvider.execute).toHaveBeenCalledOnce(); // Only once despite two calls
    }).pipe(Effect.provide(makeIdempotencySetLive()))
  );
});
```

**Step 2: Run test to verify it fails**

Run: `pnpm vitest run test/unit/provider/orchestration-effect.test.ts`
Expected: FAIL — module not found

**Step 3: Write implementation**

```typescript
// src/lib/effect/orchestration-service.ts
import { Context, Effect, Layer, Ref, Schedule, Duration } from "effect";

// Idempotency entries have a timestamp for TTL-based eviction.
// Without eviction, the set grows unboundedly with every command.
interface IdempotencyEntry {
  id: string;
  addedAt: number;
}

// AUDIT FIX (H2): Uses native Map (not HashMap) because IdempotencyEntry
// contains timestamps used in TTL comparisons. Native Map's iteration order
// is insertion-order, which simplifies TTL eviction. No Fiber or mutable
// Effect primitives, but the eviction pattern benefits from native Map.
interface IdempotencyState {
  entries: Map<string, IdempotencyEntry>;
}

export class IdempotencySetTag extends Context.Tag("IdempotencySet")<
  IdempotencySetTag,
  Ref.Ref<IdempotencyState>
>() {}

const IDEMPOTENCY_TTL_MS = 5 * 60 * 1000; // 5 minutes
const EVICTION_INTERVAL = Duration.minutes(1);

export const makeIdempotencySetLive = (): Layer.Layer<IdempotencySetTag> =>
  Layer.scoped(
    IdempotencySetTag,
    Effect.gen(function* () {
      const ref = yield* Ref.make<IdempotencyState>({ entries: new Map() });

      // Background eviction fiber — purge entries older than TTL
      yield* Effect.sync(() => Date.now()).pipe(
        Effect.flatMap((now) =>
          Ref.update(ref, (s) => {
            const entries = new Map<string, IdempotencyEntry>();
            for (const [id, entry] of s.entries) {
              if (now - entry.addedAt < IDEMPOTENCY_TTL_MS) {
                entries.set(id, entry);
              }
            }
            return { entries };
          })
        ),
        Effect.repeat(Schedule.spaced(EVICTION_INTERVAL)),
        Effect.forkScoped
      );

      return ref;
    })
  );

interface Command {
  id: string;
  type: string;
  payload: unknown;
}

interface Provider {
  execute: (cmd: Command) => Effect.Effect<any>;
}

export const routeCommand = (cmd: Command, provider: Provider) =>
  Effect.gen(function* () {
    const seenRef = yield* IdempotencySetTag;
    const now = Date.now();

    const isDuplicate = yield* Ref.modify(seenRef, (s) => {
      if (s.entries.has(cmd.id)) return [true, s];
      const entries = new Map(s.entries);
      entries.set(cmd.id, { id: cmd.id, addedAt: now });
      return [false, { entries }];
    });

    if (isDuplicate) return { deduplicated: true };

    return yield* provider.execute(cmd);
  });
```

**Step 4: Run test to verify it passes**

Run: `pnpm vitest run test/unit/provider/orchestration-effect.test.ts`
Expected: 2 tests PASS

**Step 5: Commit**

```bash
git add src/lib/effect/orchestration-service.ts test/unit/provider/orchestration-effect.test.ts
git commit -m "feat(effect): complete OrchestrationEngine migration to pure Effect"
```

---

## Task 17: PTY Upstream — Stream over WebSocket

**Files:**
- Create: `src/lib/effect/pty-stream.ts`
- Test: `test/unit/relay/pty-stream-effect.test.ts`

**Step 1: Write the failing test**

```typescript
// test/unit/relay/pty-stream-effect.test.ts
import { describe, it, expect } from "vitest";
import { Effect, Stream, Chunk, Duration, Exit } from "effect";
import { PtyConnectionTimeout, PtyConnectionError } from "../../../src/lib/effect/pty-stream.js";

describe("PTY Stream Effect", () => {
  it("PtyConnectionTimeout has correct tag", () => {
    const err = new PtyConnectionTimeout({});
    expect(err._tag).toBe("PtyConnectionTimeout");
  });

  it("PtyConnectionError wraps cause", () => {
    const err = new PtyConnectionError({ cause: new Error("ws failed") });
    expect(err._tag).toBe("PtyConnectionError");
  });

  it.effect("timeout produces PtyConnectionTimeout on stale stream", () =>
    Effect.gen(function* () {
      const neverStream = Stream.never as Stream.Stream<any, never>;
      const exit = yield* Effect.exit(
        Stream.runDrain(
          neverStream.pipe(
            Stream.timeoutFail(() => new PtyConnectionTimeout(), Duration.millis(50))
          )
        )
      );
      expect(Exit.isFailure(exit)).toBe(true);
    })
  );
});
```

**Step 2: Run test to verify it fails**

Run: `pnpm vitest run test/unit/relay/pty-stream-effect.test.ts`
Expected: FAIL — module not found

**Step 3: Write implementation**

```typescript
// src/lib/effect/pty-stream.ts
import { Data, Effect, Stream, Duration } from "effect";

export interface PtyEvent {
  type: "output" | "exit" | "error";
  data: string;
}

// Use Data.TaggedError so these work with Effect.catchTag
export class PtyConnectionError extends Data.TaggedError("PtyConnectionError")<{
  cause: unknown;
}> {}

export class PtyConnectionTimeout extends Data.TaggedError("PtyConnectionTimeout")<{}> {}

// Use top-level import — `ws` is already a project dependency.
// Do NOT use `await import("ws")` inside Effect.gen — `await` is
// invalid in generator functions that use `yield*`.
import WebSocket from "ws";

export const ptyStream = (url: string) =>
  Stream.asyncScoped<PtyEvent, PtyConnectionError | PtyConnectionTimeout>((emit) =>
    Effect.gen(function* () {
      const ws = yield* Effect.acquireRelease(
        Effect.sync(() => new WebSocket(url, {
          perMessageDeflate: false,
        })),
        (ws) => Effect.sync(() => { try { ws.close(); } catch {} })
      );

      ws.on("message", (data: Buffer) => {
        try {
          const event = JSON.parse(data.toString()) as PtyEvent;
          emit.single(event);
        } catch {
          emit.single({ type: "output", data: data.toString() });
        }
      });

      ws.on("error", (err: Error) => {
        emit.fail(new PtyConnectionError({ cause: err }));
      });

      ws.on("close", () => {
        emit.end();
      });

      // Connection timeout — fire-and-forget fiber
      yield* Effect.sleep(Duration.seconds(10)).pipe(
        Effect.flatMap(() =>
          ws.readyState !== WebSocket.OPEN
            ? Effect.sync(() => emit.fail(new PtyConnectionTimeout({})))
            : Effect.void
        ),
        Effect.forkScoped
      );
    })
  );
```

**Step 4: Run test to verify it passes**

Run: `pnpm vitest run test/unit/relay/pty-stream-effect.test.ts`
Expected: 3 tests PASS

**Step 5: Commit**

```bash
git add src/lib/effect/pty-stream.ts test/unit/relay/pty-stream-effect.test.ts
git commit -m "feat(effect): add PTY upstream as Effect.Stream over WebSocket"
```

---

## Task 18: Push Notifications — bounded concurrency + acquireRelease

> **AUDIT FIX (M8):** Title previously said "Pool" but `Effect.Pool` is for
> managing reusable pooled resources (DB connections, HTTP clients). Push
> notification sends are fire-and-forget — `Effect.forEach` with bounded
> `concurrency: 10` is the correct pattern here, not Pool.

**Files:**
- Create: `src/lib/effect/push-service.ts`
- Test: `test/unit/server/push-effect.test.ts`

**Step 1: Write the failing test**

```typescript
// test/unit/server/push-effect.test.ts
import { describe, it } from "@effect/vitest";
import { expect, vi } from "vitest";
import { Effect, Layer, Ref, Exit } from "effect";
import {
  PushManagerTag,
  PushManagerLive,
  type PushSubscription,
} from "../../../src/lib/effect/push-service.js";

describe("Push Notifications Effect", () => {
  const mockSendPush = vi.fn().mockReturnValue(Effect.succeed(undefined));

  it.scoped("subscribe adds subscription", () =>
    Effect.gen(function* () {
      const sub: PushSubscription = { id: "sub-1", endpoint: "https://push.example.com", keys: {} as any };

      const push = yield* PushManagerTag;
      yield* push.subscribe(sub);
    }).pipe(Effect.provide(PushManagerLive({ sendPush: mockSendPush })))
  );

  it.scoped("broadcast sends to all subscriptions", () =>
    Effect.gen(function* () {
      const sub1: PushSubscription = { id: "sub-1", endpoint: "https://a.com", keys: {} as any };
      const sub2: PushSubscription = { id: "sub-2", endpoint: "https://b.com", keys: {} as any };

      const push = yield* PushManagerTag;
      yield* push.subscribe(sub1);
      yield* push.subscribe(sub2);
      yield* push.broadcast({ title: "Test", body: "Hello" });

      expect(mockSendPush).toHaveBeenCalledTimes(2);
    }).pipe(Effect.provide(PushManagerLive({ sendPush: mockSendPush })))
  );

  it.scoped("individual send failure does not block others", () =>
    Effect.gen(function* () {
      const failingSend = vi.fn()
        .mockReturnValueOnce(Effect.fail(new Error("network error")))
        .mockReturnValueOnce(Effect.succeed(undefined));

      const sub1: PushSubscription = { id: "sub-1", endpoint: "https://a.com", keys: {} as any };
      const sub2: PushSubscription = { id: "sub-2", endpoint: "https://b.com", keys: {} as any };

      const exit = yield* Effect.exit(
        Effect.gen(function* () {
          const push = yield* PushManagerTag;
          yield* push.subscribe(sub1);
          yield* push.subscribe(sub2);
          yield* push.broadcast({ title: "Test", body: "Hello" });
        }).pipe(Effect.provide(PushManagerLive({ sendPush: failingSend })))
      );

      // Should succeed despite first send failing
      expect(Exit.isSuccess(exit)).toBe(true);
      expect(failingSend).toHaveBeenCalledTimes(2);
    })
  );
});
```

**Step 2: Run test to verify it fails**

Run: `pnpm vitest run test/unit/server/push-effect.test.ts`
Expected: FAIL — module not found

**Step 3: Write implementation**

```typescript
// src/lib/effect/push-service.ts
import { Context, Effect, Layer, Ref } from "effect";

export interface PushSubscription {
  id: string;
  endpoint: string;
  keys: { p256dh: string; auth: string };
}

interface PushPayload {
  title: string;
  body: string;
}

interface PushService {
  subscribe: (sub: PushSubscription) => Effect.Effect<void>;
  unsubscribe: (id: string) => Effect.Effect<void>;
  broadcast: (payload: PushPayload) => Effect.Effect<void>;
}

export class PushManagerTag extends Context.Tag("PushManager")<
  PushManagerTag,
  PushService
>() {}

interface PushManagerConfig {
  sendPush: (sub: PushSubscription, payload: PushPayload) => Effect.Effect<void, any>;
}

export const PushManagerLive = (config: PushManagerConfig) =>
  Layer.scoped(
    PushManagerTag,
    Effect.gen(function* () {
      // AUDIT FIX (H3): Uses native Map (not HashMap) because subscription
      // entries are iterated for broadcast and the Map's .values() is more
      // ergonomic. No mutable Effect primitives in values. Documented exception.
      const subscriptions = yield* Ref.make<Map<string, PushSubscription>>(new Map());

      return {
        subscribe: (sub: PushSubscription) =>
          Ref.update(subscriptions, (m) => {
            const next = new Map(m);
            next.set(sub.id, sub);
            return next;
          }),

        unsubscribe: (id: string) =>
          Ref.update(subscriptions, (m) => {
            const next = new Map(m);
            next.delete(id);
            return next;
          }),

        broadcast: (payload: PushPayload) =>
          Effect.gen(function* () {
            const subs = yield* Ref.get(subscriptions);
            // Bounded concurrency acts as a pragmatic send pool.
            // Individual failures are isolated — one bad subscription
            // doesn't block sends to others. Failed subscriptions are
            // logged but not removed (purgeDeadSubscriptions handles that).
            yield* Effect.forEach(
              [...subs.values()],
              (sub) => config.sendPush(sub, payload).pipe(
                Effect.catchAll((e) =>
                  Effect.logWarning(`Push send failed for ${sub.id}`, e)
                )
              ),
              { concurrency: 10 }
            );
          }),
      };
    })
  );
```

**Step 4: Run test to verify it passes**

Run: `pnpm vitest run test/unit/server/push-effect.test.ts`
Expected: 3 tests PASS

**Step 5: Commit**

```bash
git add src/lib/effect/push-service.ts test/unit/server/push-effect.test.ts
git commit -m "feat(effect): add push notification service with isolated send failures"
```
