"note": "Heuristic false-green LEADS, not findings. A finding requires a demonstrated reproduction (surviving mutation via mutation-runner). Do not treat a candidate as a confirmed false green. Coverage-theater leads may include tests that delegate their assertion to a custom helper the classifier does not recognise; verify before acting.",

  "sourceDigest": "sha256:dd810dd48c5bdbc9becd7fcc01dd41a4ca2abf0b2d6f6a545907247f6e3e8361",

  "summary": {

    "filesScanned": 2468,

    "filesScanned": 2469,

    "candidateCount": 16,

    "byMode": {

      "false_green_coverage_theater": 15,

{

  "schemaVersion": "1",

  "repository": "OpenAgentsInc/openagents",

  "sourceDigest": "sha256:359b6533e42f95b7605e98f21e6ed7a4f837f917a73ad1723aab8cee17ec0752",

  "sourceDigest": "sha256:d6a28da451ce9ef6f5cf228f613eb7a4a8e932087110a750f3bb6d198737304c",

  "surfaces": [

    {

      "id": "app:@openagentsinc/acceptance-runner",

      "oracles": [

        {

          "type": "test",

          "ref": "packages/agent-experience-memory (10 tracked test files)"

          "ref": "packages/agent-experience-memory (11 tracked test files)"

        }

      ],

      "obligation": {

export * from "./consolidation.js";

export * from "./subagent-memory.js";

export * from "./projection.js";

export * from "./sync.js";

import { describe, expect, test } from "vite-plus/test";

import {

  buildEngramBody,

  buildEngramEvent,

  engramContentDigest,

  type EngramEvent,

} from "./engram.js";

import {

  EngramSyncQueue,

  MemoryTransport,

  type EngramTransport,

  type PublishResult,

} from "./sync.js";

const PUBKEY = "a".repeat(64);

const sign = (eventId: string): string => `sig-${eventId.slice(0, 16)}`;

const engram = (slug: string, value: string, createdAt: number): EngramEvent => {

  const body = buildEngramBody(slug, value, {

    admission: "admitted",

    entityId: "entity-1",

    contentDigest: engramContentDigest(value),

    sourceEventRefs: [],

    relations: [],

    derivedFromSlugs: [],

  });

  return buildEngramEvent(PUBKEY, createdAt, slug, JSON.stringify(body), sign);

};

describe("the memory transport", () => {

  test("stores what it accepts and finds it by filter", async () => {

    const transport = new MemoryTransport();

    const first = engram("note/one", "first", 1_000);

    const second = engram("note/two", "second", 2_000);

    await transport.publish(first);

    await transport.publish(second);

    expect(await transport.fetch({ slugs: ["note/one"] })).toEqual([first]);

    expect(await transport.fetch({ since: 1_500 })).toEqual([second]);

    expect(await transport.fetch({ authors: [PUBKEY] })).toHaveLength(2);

    expect(await transport.fetch({ limit: 1 })).toEqual([first]);

  });

  test("refuses an engram whose id does not verify", async () => {

    const transport = new MemoryTransport();

    const authentic = engram("note/one", "authentic", 1_000);

    const tampered: EngramEvent = {

      ...authentic,

      content: authentic.content.replace("authentic", "forged"),

    };

    const result = await transport.publish(tampered);

    expect(result).toEqual({ ok: false, reason: "refused", detail: "event id does not verify" });

    expect(transport.stored()).toHaveLength(0);

  });

});

describe("the sync queue", () => {

  test("publishing does not wait on the transport", () => {

    const queue = new EngramSyncQueue(new MemoryTransport());

    // No await: publish returns nothing to wait on, and the engram is queued.

    queue.publish(engram("note/one", "first", 1_000));

    expect(queue.status().pending).toBe(1);

    expect(queue.status().delivered).toBe(0);

  });

  test("a drain delivers what is queued", async () => {

    const transport = new MemoryTransport();

    const queue = new EngramSyncQueue(transport);

    queue.publish(engram("note/one", "first", 1_000));

    queue.publish(engram("note/two", "second", 2_000));

    expect(await queue.drain()).toBe(2);

    expect(queue.status()).toMatchObject({ pending: 0, delivered: 2, refused: 0 });

    expect(transport.stored()).toHaveLength(2);

  });

  test("an unreachable transport loses nothing and says it is behind", async () => {

    const transport = new MemoryTransport();

    transport.reachable = false;

    const queue = new EngramSyncQueue(transport);

    queue.publish(engram("note/one", "first", 1_000));

    expect(await queue.drain()).toBe(0);

    expect(queue.behind()).toBe(true);

    expect(queue.status()).toMatchObject({ pending: 1, delivered: 0 });

    expect(queue.status().lastFailure?.reason).toBe("unreachable");

    // The engram is still there when the transport comes back.

    transport.reachable = true;

    expect(await queue.drain()).toBe(1);

    expect(queue.behind()).toBe(false);

    expect(transport.stored()).toHaveLength(1);

  });

  test("a refusal is terminal and does not retry forever", async () => {

    const transport = new MemoryTransport();

    const queue = new EngramSyncQueue(transport);

    const authentic = engram("note/one", "authentic", 1_000);

    queue.publish({ ...authentic, content: authentic.content.replace("authentic", "forged") });

    expect(await queue.drain()).toBe(0);

    expect(queue.status()).toMatchObject({ pending: 0, refused: 1 });

    // Nothing left to retry: the transport judged it, and repeating the call

    // only repeats the judgement.

    expect(await queue.drain()).toBe(0);

    expect(queue.behind()).toBe(false);

  });

  test("a transport that throws is treated as failing, not as a crash", async () => {

    const throwing: EngramTransport = {

      publish(): Promise<PublishResult> {

        throw new Error("socket exploded");

      },

      fetch(): Promise<ReadonlyArray<EngramEvent>> {

        throw new Error("socket exploded");

      },

    };

    const queue = new EngramSyncQueue(throwing);

    queue.publish(engram("note/one", "first", 1_000));

    await expect(queue.drain()).resolves.toBe(0);

    expect(queue.status().lastFailure).toMatchObject({

      reason: "failed",

      detail: "socket exploded",

    });

    expect(queue.status().pending).toBe(1);

    await expect(queue.fetch({})).resolves.toEqual([]);

  });

  test("an engram is never delivered twice", async () => {

    const transport = new MemoryTransport();

    const queue = new EngramSyncQueue(transport);

    const event = engram("note/one", "first", 1_000);

    queue.publish(event);

    expect(await queue.drain()).toBe(1);

    // Republishing the whole ledger costs one pass, not one delivery per pass.

    queue.publish(event);

    expect(queue.status().pending).toBe(0);

    expect(await queue.drain()).toBe(0);

    expect(queue.status().delivered).toBe(1);

  });

  test("the same engram queued twice before a drain delivers once", async () => {

    const transport = new MemoryTransport();

    const queue = new EngramSyncQueue(transport);

    const event = engram("note/one", "first", 1_000);

    queue.publish(event);

    queue.publish(event);

    expect(queue.status().pending).toBe(1);

    expect(await queue.drain()).toBe(1);

  });

  test("fetching reads back what was delivered", async () => {

    const transport = new MemoryTransport();

    const queue = new EngramSyncQueue(transport);

    const event = engram("note/one", "first", 1_000);

    queue.publish(event);

    await queue.drain();

    expect(await queue.fetch({ slugs: ["note/one"] })).toEqual([event]);

  });

});

import { Schema as S } from "effect";

import { computeEngramEventId, type EngramEvent } from "./engram.js";

/**

 * The engram sync seam (issue #222).

 *

 * The ledger is local and authoritative. Sync is how a copy of it reaches

 * somewhere else, and the whole design follows from one rule: **a turn never

 * waits for it, and never fails because of it.** A relay that is slow, down,

 * or gone must be indistinguishable from one that is fine, as far as the

 * caller is concerned — because memory that can break a conversation is worse

 * than memory that is briefly out of date.

 *

 * That gives the three properties the tests hold this to:

 *

 * 1. **Local-first.** `publish` records the engram as pending and returns

 *    immediately. The engram is already in the local ledger by then; sync is

 *    catching up, not gatekeeping.

 * 2. **Nothing is lost.** A failed publish stays queued and is retried. The

 *    queue only forgets an engram once a transport has acknowledged it, so a

 *    transport that is down for an hour costs an hour of latency and no

 *    engrams.

 * 3. **Degraded is a state, not an error.** The queue reports what it is

 *    holding and what failed last, so a caller can *say* it is behind rather

 *    than discovering it by silence.

 *

 * What is deliberately not here: relays, sockets, keys, encryption. A

 * transport is anything that satisfies `EngramTransport`, and the real Nostr

 * one lands behind this interface without any caller changing. The workspace

 * has a shared Nostr implementation (`nostr-effect`) that a real transport

 * should build on rather than reimplement.

 */

export const SYNC_SCHEMA_ID = "openagents.engram_sync.v1" as const;

/** Why a publish did not land. Distinct because they need distinct responses. */

export const SyncFailureReason = S.Literals([

  /** The transport could not be reached at all. Retry later, unchanged. */

  "unreachable",

  /** The transport reached and refused this engram. Retrying will not help. */

  "refused",

  /** The transport accepted the call and failed inside it. Retry later. */

  "failed",

]);

export type SyncFailureReason = typeof SyncFailureReason.Type;

export type PublishResult =

  | { readonly ok: true; readonly eventId: string }

  | { readonly ok: false; readonly reason: SyncFailureReason; readonly detail?: string };

/** A filter for what to fetch back. Absent fields mean "no constraint". */

export interface EngramFilter {

  readonly authors?: ReadonlyArray<string>;

  /** `d` tag values — engram slugs. */

  readonly slugs?: ReadonlyArray<string>;

  /** Unix seconds; inclusive. */

  readonly since?: number;

  readonly until?: number;

  readonly limit?: number;

}

/**

 * Whatever carries engrams somewhere else.

 *

 * Implementations must not throw: a transport that cannot answer returns a

 * failure, because a throw from here would reach the caller's turn and the

 * whole point is that it cannot.

 */

export interface EngramTransport {

  publish(event: EngramEvent): Promise<PublishResult>;

  fetch(filter: EngramFilter): Promise<ReadonlyArray<EngramEvent>>;

}

/** What the queue is holding, for a caller that wants to say so. */

export interface SyncStatus {

  readonly pending: number;

  readonly delivered: number;

  /** Engrams the transport refused outright. They are not retried. */

  readonly refused: number;

  readonly lastFailure?: { readonly reason: SyncFailureReason; readonly detail?: string };

}

const matches = (event: EngramEvent, filter: EngramFilter): boolean => {

  if (filter.authors !== undefined && !filter.authors.includes(event.pubkey)) return false;

  if (filter.since !== undefined && event.created_at < filter.since) return false;

  if (filter.until !== undefined && event.created_at > filter.until) return false;

  if (filter.slugs !== undefined) {

    const slug = event.tags.find((tag) => tag[0] === "d")?.[1];

    if (slug === undefined || !filter.slugs.includes(slug)) return false;

  }

  return true;

};

/**

 * A transport that keeps engrams in memory.

 *

 * The reference implementation and what the tests run against. It is also a

 * useful real transport for a single process that wants sync's shape without

 * a relay.

 */

export class MemoryTransport implements EngramTransport {

  private readonly events = new Map<string, EngramEvent>();

  /** Set to fail every publish, for degraded-mode tests. */

  reachable = true;

  publish(event: EngramEvent): Promise<PublishResult> {

    if (!this.reachable) {

      return Promise.resolve({ ok: false, reason: "unreachable", detail: "transport is down" });

    }

    if (event.id !== computeEngramEventId(event)) {

      return Promise.resolve({ ok: false, reason: "refused", detail: "event id does not verify" });

    }

    this.events.set(event.id, event);

    return Promise.resolve({ ok: true, eventId: event.id });

  }

  fetch(filter: EngramFilter): Promise<ReadonlyArray<EngramEvent>> {

    if (!this.reachable) return Promise.resolve([]);

    const found = [...this.events.values()]

      .filter((event) => matches(event, filter))

      .sort((left, right) =>

        left.created_at !== right.created_at

          ? left.created_at - right.created_at

          : left.id.localeCompare(right.id),

      );

    return Promise.resolve(filter.limit === undefined ? found : found.slice(0, filter.limit));

  }

  /** Everything the transport holds, for assertions. */

  stored(): ReadonlyArray<EngramEvent> {

    return [...this.events.values()];

  }

}

/**

 * The queue between the ledger and a transport.

 *

 * `publish` is synchronous from the caller's side: it enqueues and returns.

 * `drain` is what actually talks to the transport, and a host calls it

 * whenever it likes — after a turn, on a timer, at exit. Nothing about the

 * caller's turn depends on when that happens.

 */

export class EngramSyncQueue {

  private readonly queue: Array<EngramEvent> = [];

  private readonly deliveredIds = new Set<string>();

  private readonly refusedIds = new Set<string>();

  private lastFailure: SyncStatus["lastFailure"];

  private draining = false;

  constructor(private readonly transport: EngramTransport) {}

  /**

   * Enqueue an engram for delivery. Returns nothing to wait on.

   *

   * An engram already delivered or already refused is not enqueued twice, so

   * a caller that republishes its whole ledger costs one pass, not one

   * delivery per pass.

   */

  publish(event: EngramEvent): void {

    if (this.deliveredIds.has(event.id) || this.refusedIds.has(event.id)) return;

    if (this.queue.some((queued) => queued.id === event.id)) return;

    this.queue.push(event);

  }

  /**

   * Try to deliver everything queued.

   *

   * Returns how many landed. An unreachable or failing transport leaves the

   * engram queued for the next drain; a refusal is terminal, because retrying

   * something the transport has judged invalid only repeats the judgement.

   * Never throws — a transport that throws anyway is treated as failing.

   */

  async drain(): Promise<number> {

    if (this.draining) return 0;

    this.draining = true;

    try {

      let delivered = 0;

      // Copy: a publish during a drain lands in the next one rather than

      // mutating the array being walked.

      const attempting = [...this.queue];

      for (const event of attempting) {

        let result: PublishResult;

        try {

          result = await this.transport.publish(event);

        } catch (cause) {

          result = {

            ok: false,

            reason: "failed",

            detail: cause instanceof Error ? cause.message : String(cause),

          };

        }

        if (result.ok) {

          this.deliveredIds.add(event.id);

          this.removeFromQueue(event.id);

          delivered += 1;

          continue;

        }

        this.lastFailure = {

          reason: result.reason,

          ...(result.detail === undefined ? {} : { detail: result.detail }),

        };

        if (result.reason === "refused") {

          this.refusedIds.add(event.id);

          this.removeFromQueue(event.id);

        }

        // "unreachable" and "failed" stay queued for the next drain.

      }

      return delivered;

    } finally {

      this.draining = false;

    }

  }

  private removeFromQueue(eventId: string): void {

    const at = this.queue.findIndex((queued) => queued.id === eventId);

    if (at >= 0) this.queue.splice(at, 1);

  }

  /** Fetch from the transport. An unreachable transport yields nothing. */

  async fetch(filter: EngramFilter): Promise<ReadonlyArray<EngramEvent>> {

    try {

      return await this.transport.fetch(filter);

    } catch {

      return [];

    }

  }

  status(): SyncStatus {

    return {

      pending: this.queue.length,

      delivered: this.deliveredIds.size,

      refused: this.refusedIds.size,

      ...(this.lastFailure === undefined ? {} : { lastFailure: this.lastFailure }),

    };

  }

  /** Whether the queue is holding anything the transport has not taken. */

  behind(): boolean {

    return this.queue.length > 0;

  }

}

      ["./internal/sha256.js", "./sha256.js"],

    ],

  ],

  ["agent-experience-memory/src/sync.ts", "sync.ts", []],

  [

    "agent-experience-memory/src/consolidation.ts",

    "consolidation.ts",

export * from "./consolidation.js";

export * from "./subagent-memory.js";

export * from "./projection.js";

export * from "./sync.js";

`;

export const writeAll = () => {

  ledgerEntriesAsHeuristics,

  promoteHeuristicToPattern,

  signSupersedingEngram,

  EngramSyncQueue,

  MemoryTransport,

  project,

  projectedValue,

  projectionMatches,

  HarvestedLedgerEntry,

  type EngramEvent,

  type EngramBody,

  type EngramTransport,

  type ParentHeuristic,

  type Projection,

  type SyncStatus,

} from "./memory/index.js";

import { Schema as S } from "effect";

  readonly projectScope?: string;

  /** Epoch-milliseconds clock, injectable for tests. */

  readonly now?: () => number;

  /**

   * Where engrams are mirrored. Defaults to an in-process transport, which

   * keeps the seam exercised without a relay; a real Nostr transport drops in

   * here without any caller changing (#222).

   */

  readonly transport?: EngramTransport;

  /**

   * New harvests needed before the harvest path schedules a dream.

   * `Infinity` turns the automatic pass off, leaving `dream` explicit.

  private readonly dreamThreshold: number;

  private key: Buffer | undefined;

  private cachedProjection: Projection | undefined;

  private readonly sync: EngramSyncQueue;

  constructor(options: CoderMemoryOptions = {}) {

    this.directory = options.directory ?? join(homedir(), ".openagents", "memory");

    this.projectScope = sanitizeScope(options.projectScope ?? `project:${process.cwd()}`);

    this.now = options.now ?? Date.now;

    this.dreamThreshold = options.dreamThreshold ?? DREAM_THRESHOLD;

    this.sync = new EngramSyncQueue(options.transport ?? new MemoryTransport());

  }

  /**

   * What sync is holding. A session can say it is behind rather than leaving

   * a reader to infer it from silence.

   */

  syncStatus(): SyncStatus {

    return this.sync.status();

  }

  /**

   * Hand everything queued to the transport.

   *

   * Called between turns, never inside one: the local ledger is already

   * authoritative by the time this runs, so a transport that is down costs

   * latency and nothing else.

   */

  async flush(): Promise<number> {

    return this.sync.drain();

  }

  /** The local signing key, created on first use. */

    );

    appendFileSync(this.ledgerPath, `${JSON.stringify(event)}\n`, { mode: 0o600 });

    this.cachedProjection = undefined;

    // Local-first: the engram is on disk and readable now. Sync catches up.

    this.sync.publish(event);

    return event;

  }

      sign,

    );

    appendFileSync(this.ledgerPath, `${JSON.stringify(event)}\n`, { mode: 0o600 });

    this.cachedProjection = undefined;

    this.sync.publish(event);

    return event;

  }

export * from "./consolidation.js";

export * from "./subagent-memory.js";

export * from "./projection.js";

export * from "./sync.js";

// Vendored from packages/agent-experience-memory/src/sync.ts by scripts/vendor-memory.mjs — do not edit here.

// The drift guard (test/vendored-memory-drift.test.ts) fails when this copy

// no longer matches the canonical source.

import { Schema as S } from "effect";

import { computeEngramEventId, type EngramEvent } from "./engram.js";

/**

 * The engram sync seam (issue #222).

 *

 * The ledger is local and authoritative. Sync is how a copy of it reaches

 * somewhere else, and the whole design follows from one rule: **a turn never

 * waits for it, and never fails because of it.** A relay that is slow, down,

 * or gone must be indistinguishable from one that is fine, as far as the

 * caller is concerned — because memory that can break a conversation is worse

 * than memory that is briefly out of date.

 *

 * That gives the three properties the tests hold this to:

 *

 * 1. **Local-first.** `publish` records the engram as pending and returns

 *    immediately. The engram is already in the local ledger by then; sync is

 *    catching up, not gatekeeping.

 * 2. **Nothing is lost.** A failed publish stays queued and is retried. The

 *    queue only forgets an engram once a transport has acknowledged it, so a

 *    transport that is down for an hour costs an hour of latency and no

 *    engrams.

 * 3. **Degraded is a state, not an error.** The queue reports what it is

 *    holding and what failed last, so a caller can *say* it is behind rather

 *    than discovering it by silence.

 *

 * What is deliberately not here: relays, sockets, keys, encryption. A

 * transport is anything that satisfies `EngramTransport`, and the real Nostr

 * one lands behind this interface without any caller changing. The workspace

 * has a shared Nostr implementation (`nostr-effect`) that a real transport

 * should build on rather than reimplement.

 */

export const SYNC_SCHEMA_ID = "openagents.engram_sync.v1" as const;

/** Why a publish did not land. Distinct because they need distinct responses. */

export const SyncFailureReason = S.Literals([

  /** The transport could not be reached at all. Retry later, unchanged. */

  "unreachable",

  /** The transport reached and refused this engram. Retrying will not help. */

  "refused",

  /** The transport accepted the call and failed inside it. Retry later. */

  "failed",

]);

export type SyncFailureReason = typeof SyncFailureReason.Type;

export type PublishResult =

  | { readonly ok: true; readonly eventId: string }

  | { readonly ok: false; readonly reason: SyncFailureReason; readonly detail?: string };

/** A filter for what to fetch back. Absent fields mean "no constraint". */

export interface EngramFilter {

  readonly authors?: ReadonlyArray<string>;

  /** `d` tag values — engram slugs. */

  readonly slugs?: ReadonlyArray<string>;

  /** Unix seconds; inclusive. */

  readonly since?: number;

  readonly until?: number;

  readonly limit?: number;

}

/**

 * Whatever carries engrams somewhere else.

 *

 * Implementations must not throw: a transport that cannot answer returns a

 * failure, because a throw from here would reach the caller's turn and the

 * whole point is that it cannot.

 */

export interface EngramTransport {

  publish(event: EngramEvent): Promise<PublishResult>;

  fetch(filter: EngramFilter): Promise<ReadonlyArray<EngramEvent>>;

}

/** What the queue is holding, for a caller that wants to say so. */

export interface SyncStatus {

  readonly pending: number;

  readonly delivered: number;

  /** Engrams the transport refused outright. They are not retried. */

  readonly refused: number;

  readonly lastFailure?: { readonly reason: SyncFailureReason; readonly detail?: string };

}

const matches = (event: EngramEvent, filter: EngramFilter): boolean => {

  if (filter.authors !== undefined && !filter.authors.includes(event.pubkey)) return false;

  if (filter.since !== undefined && event.created_at < filter.since) return false;

  if (filter.until !== undefined && event.created_at > filter.until) return false;

  if (filter.slugs !== undefined) {

    const slug = event.tags.find((tag) => tag[0] === "d")?.[1];

    if (slug === undefined || !filter.slugs.includes(slug)) return false;

  }

  return true;

};

/**

 * A transport that keeps engrams in memory.

 *

 * The reference implementation and what the tests run against. It is also a

 * useful real transport for a single process that wants sync's shape without

 * a relay.

 */

export class MemoryTransport implements EngramTransport {

  private readonly events = new Map<string, EngramEvent>();

  /** Set to fail every publish, for degraded-mode tests. */

  reachable = true;

  publish(event: EngramEvent): Promise<PublishResult> {

    if (!this.reachable) {

      return Promise.resolve({ ok: false, reason: "unreachable", detail: "transport is down" });

    }

    if (event.id !== computeEngramEventId(event)) {

      return Promise.resolve({ ok: false, reason: "refused", detail: "event id does not verify" });

    }

    this.events.set(event.id, event);

    return Promise.resolve({ ok: true, eventId: event.id });

  }

  fetch(filter: EngramFilter): Promise<ReadonlyArray<EngramEvent>> {

    if (!this.reachable) return Promise.resolve([]);

    const found = [...this.events.values()]

      .filter((event) => matches(event, filter))

      .sort((left, right) =>

        left.created_at !== right.created_at

          ? left.created_at - right.created_at

          : left.id.localeCompare(right.id),

      );

    return Promise.resolve(filter.limit === undefined ? found : found.slice(0, filter.limit));

  }

  /** Everything the transport holds, for assertions. */

  stored(): ReadonlyArray<EngramEvent> {

    return [...this.events.values()];

  }

}

/**

 * The queue between the ledger and a transport.

 *

 * `publish` is synchronous from the caller's side: it enqueues and returns.

 * `drain` is what actually talks to the transport, and a host calls it

 * whenever it likes — after a turn, on a timer, at exit. Nothing about the

 * caller's turn depends on when that happens.

 */

export class EngramSyncQueue {

  private readonly queue: Array<EngramEvent> = [];

  private readonly deliveredIds = new Set<string>();

  private readonly refusedIds = new Set<string>();

  private lastFailure: SyncStatus["lastFailure"];

  private draining = false;

  constructor(private readonly transport: EngramTransport) {}

  /**

   * Enqueue an engram for delivery. Returns nothing to wait on.

   *

   * An engram already delivered or already refused is not enqueued twice, so

   * a caller that republishes its whole ledger costs one pass, not one

   * delivery per pass.

   */

  publish(event: EngramEvent): void {

    if (this.deliveredIds.has(event.id) || this.refusedIds.has(event.id)) return;

    if (this.queue.some((queued) => queued.id === event.id)) return;

    this.queue.push(event);

  }

  /**

   * Try to deliver everything queued.

   *

   * Returns how many landed. An unreachable or failing transport leaves the

   * engram queued for the next drain; a refusal is terminal, because retrying

   * something the transport has judged invalid only repeats the judgement.

   * Never throws — a transport that throws anyway is treated as failing.

   */

  async drain(): Promise<number> {

    if (this.draining) return 0;

    this.draining = true;

    try {

      let delivered = 0;

      // Copy: a publish during a drain lands in the next one rather than

      // mutating the array being walked.

      const attempting = [...this.queue];

      for (const event of attempting) {

        let result: PublishResult;

        try {

          result = await this.transport.publish(event);

        } catch (cause) {

          result = {

            ok: false,

            reason: "failed",

            detail: cause instanceof Error ? cause.message : String(cause),

          };

        }

        if (result.ok) {

          this.deliveredIds.add(event.id);

          this.removeFromQueue(event.id);

          delivered += 1;

          continue;

        }

        this.lastFailure = {

          reason: result.reason,

          ...(result.detail === undefined ? {} : { detail: result.detail }),

        };

        if (result.reason === "refused") {

          this.refusedIds.add(event.id);

          this.removeFromQueue(event.id);

        }

        // "unreachable" and "failed" stay queued for the next drain.

      }

      return delivered;

    } finally {

      this.draining = false;

    }

  }

  private removeFromQueue(eventId: string): void {

    const at = this.queue.findIndex((queued) => queued.id === eventId);

    if (at >= 0) this.queue.splice(at, 1);

  }

  /** Fetch from the transport. An unreachable transport yields nothing. */

  async fetch(filter: EngramFilter): Promise<ReadonlyArray<EngramEvent>> {

    try {

      return await this.transport.fetch(filter);

    } catch {

      return [];

    }

  }

  status(): SyncStatus {

    return {

      pending: this.queue.length,

      delivered: this.deliveredIds.size,

      refused: this.refusedIds.size,

      ...(this.lastFailure === undefined ? {} : { lastFailure: this.lastFailure }),

    };

  }

  /** Whether the queue is holding anything the transport has not taken. */

  behind(): boolean {

    return this.queue.length > 0;

  }

}

import { afterEach, describe, expect, it } from "vitest";

import { CoderMemory } from "../src/coder-memory.js";

import { MemoryTransport } from "../src/memory/index.js";

import { DelegateFleet, type DelegateEvent, type DelegateHarness } from "../src/coder-delegate.js";

import { CoderTaskRegistry } from "../src/coder-tasks.js";

    expect(distilled.length).toBeGreaterThan(0);

  });

});

describe("memory sync", () => {

  it("queues every recorded engram and delivers it on flush", async () => {

    const dir = freshDir();

    const memory = memoryAt(dir);

    memory.record("note/one", "a finding worth keeping", "note-1");

    // Local-first: the engram is readable before anything is delivered.

    expect(memory.recall("note/one")).toBe("a finding worth keeping");

    expect(memory.syncStatus().pending).toBe(1);

    expect(memory.syncStatus().delivered).toBe(0);

    await memory.flush();

    expect(memory.syncStatus()).toMatchObject({ pending: 0, delivered: 1 });

  });

  it("keeps working when the transport is down, and loses nothing", async () => {

    const dir = freshDir();

    const transport = new MemoryTransport();

    transport.reachable = false;

    const memory = new CoderMemory({

      directory: dir,

      projectScope: "project:test",

      now: () => 1_756_000_000_000,

      dreamThreshold: Number.POSITIVE_INFINITY,

      transport,

    });

    memory.record("note/one", "recorded while the relay was gone", "note-1");

    expect(memory.recall("note/one")).toBe("recorded while the relay was gone");

    await memory.flush();

    expect(memory.syncStatus().pending).toBe(1);

    expect(memory.syncStatus().lastFailure?.reason).toBe("unreachable");

    transport.reachable = true;

    await memory.flush();

    expect(memory.syncStatus()).toMatchObject({ pending: 0, delivered: 1 });

    expect(transport.stored()).toHaveLength(1);

  });

});