From 6383105c3592a24b53648b663d71fd6df6f98ef1 Mon Sep 17 00:00:00 2001 From: Volodymyr Vreshch Date: Wed, 8 Jul 2026 00:31:16 +0200 Subject: [PATCH] refactor(sync): channel-based structure, split oversized modules --- CLAUDE.md | 20 +- src/commands/vault-sync.test.ts | 4 +- src/commands/vault-sync.ts | 6 +- src/commands/vault.ts | 2 +- src/daemon-entry.ts | 2 +- src/daemon/server.ts | 4 +- src/lib/daemon-client.ts | 4 +- src/lib/vault-registry.ts | 2 +- src/sync/couch/cycle.ts | 60 +++++ src/sync/couch/local-commit.ts | 22 ++ src/sync/couch/manager.fixtures.ts | 63 +++++ src/sync/couch/manager.test.ts | 218 +---------------- src/sync/couch/manager.ts | 316 ++++--------------------- src/sync/couch/manager.types.ts | 104 ++++++++ src/sync/couch/mutation-target.test.ts | 51 ++++ src/sync/couch/mutation-target.ts | 31 +++ src/sync/couch/push-on-write.test.ts | 113 +++++++++ src/sync/couch/push-on-write.ts | 36 +++ src/sync/couch/targets.ts | 2 +- src/sync/couch/wire.ts | 41 ++++ src/sync/{ => git}/conflict.test.ts | 0 src/sync/{ => git}/conflict.ts | 0 src/sync/{ => git}/cycle.test.ts | 0 src/sync/{ => git}/cycle.ts | 0 src/sync/{ => git}/git-exec.test.ts | 0 src/sync/{ => git}/git-exec.ts | 0 src/sync/{ => git}/manager.test.ts | 0 src/sync/{ => git}/manager.ts | 6 +- src/sync/{ => git}/planner.test.ts | 0 src/sync/{ => git}/planner.ts | 0 src/sync/{ => git}/remote-url.test.ts | 0 src/sync/{ => git}/remote-url.ts | 0 vitest.config.ts | 8 +- 33 files changed, 606 insertions(+), 509 deletions(-) create mode 100644 src/sync/couch/cycle.ts create mode 100644 src/sync/couch/local-commit.ts create mode 100644 src/sync/couch/manager.fixtures.ts create mode 100644 src/sync/couch/manager.types.ts create mode 100644 src/sync/couch/mutation-target.test.ts create mode 100644 src/sync/couch/mutation-target.ts create mode 100644 src/sync/couch/push-on-write.test.ts create mode 100644 src/sync/couch/push-on-write.ts create mode 100644 src/sync/couch/wire.ts rename src/sync/{ => git}/conflict.test.ts (100%) rename src/sync/{ => git}/conflict.ts (100%) rename src/sync/{ => git}/cycle.test.ts (100%) rename src/sync/{ => git}/cycle.ts (100%) rename src/sync/{ => git}/git-exec.test.ts (100%) rename src/sync/{ => git}/git-exec.ts (100%) rename src/sync/{ => git}/manager.test.ts (100%) rename src/sync/{ => git}/manager.ts (96%) rename src/sync/{ => git}/planner.test.ts (100%) rename src/sync/{ => git}/planner.ts (100%) rename src/sync/{ => git}/remote-url.test.ts (100%) rename src/sync/{ => git}/remote-url.ts (100%) diff --git a/CLAUDE.md b/CLAUDE.md index 3214032..648ef56 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,12 +19,20 @@ its agent-runtime patterns (the local memory daemon was deliberately ported from `AGENTAGE_NO_DAEMON=1`, or fork blocked) - `src/daemon/` + `src/daemon-entry.ts` - the local daemon (node:http, 127.0.0.1 only): one in-process engine serialises vault mutations; `agentage daemon start|stop|status` -- `src/sync/` - git sync (M4): the daemon acts on per-vault `origin[]` (external remotes only), - debounced commit+push / pull-rebase per `interval` (SECONDS; 0 = manual-only). Conflicts keep - both sides (`.conflict.md` = remote copy, zero lost writes); `ignore` rides on - `.git/info/exclude` (defaults `.obsidian/` + `data.json`, a set value REPLACES them, `[]` = all). - `spawn git` directly (memory-core's createGit is private); no new deps. `vault sync [name]` - forces a cycle via the daemon (`/api/sync/run`) or in-process when it is down +- `src/sync/` - channel-based sync; one subfolder per channel, no shared root logic: + - `src/sync/git/` - git channel (M4): `manager` (scheduler + status), `cycle` (one commit+push / + pull-rebase round), `planner` (targets + interval math), `git-exec` (spawn git, error classify), + `remote-url` (allowlist + redact), `conflict` (`.conflict.md` naming). The daemon acts on + per-vault `origin[]` (external remotes only), debounced per `interval` (SECONDS; 0 = manual-only); + conflicts keep both sides (zero lost writes); `ignore` rides on `.git/info/exclude` (defaults + `.obsidian/` + `data.json`, a set value REPLACES them, `[]` = all). `spawn git` directly + (memory-core's createGit is private); no new deps + - `src/sync/couch/` - couch channel (account vaults): `manager` (thin composition), `manager.types` + (shared shapes), `cycle` / `wire` / `push-on-write` (the sync-on-save + scheduling seams), + `mutation-target` (verb -> vault+path), `local-commit`, `discovery`, `file-store`, `state-store`, + `targets` + - `src/sync/discover/` - `watcher` (auto-discovers new vaults, provisions account vaults) + - `vault sync [name]` forces a git cycle via the daemon (`/api/sync/run`) or in-process when down - `src/package-guard.test.ts` - CI guard: no agent-runtime remnants (express/ws/sqlite/ core/platform/supabase), runtime deps stay exactly `@agentage/memory-core + @agentage/server-memory + @modelcontextprotocol/sdk + chalk + commander + open` diff --git a/src/commands/vault-sync.test.ts b/src/commands/vault-sync.test.ts index 959735e..a64cac3 100644 --- a/src/commands/vault-sync.test.ts +++ b/src/commands/vault-sync.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it, vi } from 'vitest'; import { type VaultsConfig } from '@agentage/memory-core'; -import { type SyncResult } from '../sync/cycle.js'; +import { type SyncResult } from '../sync/git/cycle.js'; import { type CouchSyncResult } from '../sync/couch/manager.js'; -import { type SyncTarget } from '../sync/planner.js'; +import { type SyncTarget } from '../sync/git/planner.js'; import { runVaultSync, type VaultSyncDeps } from './vault-sync.js'; const result = (over: Partial = {}): SyncResult => ({ diff --git a/src/commands/vault-sync.ts b/src/commands/vault-sync.ts index b768d10..a5c14bf 100644 --- a/src/commands/vault-sync.ts +++ b/src/commands/vault-sync.ts @@ -4,11 +4,11 @@ import { health, syncRun, type SyncRunResult } from '../lib/daemon-client.js'; import { daemonDisabled } from '../lib/daemon-pref.js'; import { loadVaultsConfig } from '../lib/vaults.js'; import { resolvePort } from '../daemon/lifecycle.js'; -import { runSyncCycle, type SyncResult } from '../sync/cycle.js'; +import { runSyncCycle, type SyncResult } from '../sync/git/cycle.js'; import { createCouchSyncManager, type CouchSyncResult } from '../sync/couch/manager.js'; import { couchTargets } from '../sync/couch/targets.js'; -import { syncTargets, type SyncTarget } from '../sync/planner.js'; -import { redactRemoteUrl } from '../sync/remote-url.js'; +import { syncTargets, type SyncTarget } from '../sync/git/planner.js'; +import { redactRemoteUrl } from '../sync/git/remote-url.js'; export interface VaultSyncDeps { loadConfig: () => VaultsConfig; diff --git a/src/commands/vault.ts b/src/commands/vault.ts index 24f988e..a56fa39 100644 --- a/src/commands/vault.ts +++ b/src/commands/vault.ts @@ -16,7 +16,7 @@ import { type ProvisionResult, } from '../lib/provision.js'; import { loadVaultsConfig, mutateVaultsConfig, type LoadedVaults } from '../lib/vaults.js'; -import { assertSafeRemoteUrl, redactRemoteUrl } from '../sync/remote-url.js'; +import { assertSafeRemoteUrl, redactRemoteUrl } from '../sync/git/remote-url.js'; import { defaultVaultSyncDeps, runVaultSync } from './vault-sync.js'; export interface VaultDeps { diff --git a/src/daemon-entry.ts b/src/daemon-entry.ts index 68d7186..ab0b556 100644 --- a/src/daemon-entry.ts +++ b/src/daemon-entry.ts @@ -18,7 +18,7 @@ import { loadLocalMemoryServer } from './mcp/local-server.js'; import { loadVaultsConfig, vaultsJsonPath } from './lib/vaults.js'; import { createCouchSyncManager } from './sync/couch/manager.js'; import { createDiscoverWatcher } from './sync/discover/watcher.js'; -import { createSyncManager } from './sync/manager.js'; +import { createSyncManager } from './sync/git/manager.js'; import { VERSION } from './utils/version.js'; export const isEaddrinuse = (err: unknown): boolean => diff --git a/src/daemon/server.ts b/src/daemon/server.ts index ae6842f..b345013 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -1,9 +1,9 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'; import { type McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; import { type MemoryClient } from '../lib/memory-client.js'; -import { type SyncResult } from '../sync/cycle.js'; +import { type SyncResult } from '../sync/git/cycle.js'; import { type CouchSyncResult } from '../sync/couch/manager.js'; -import { type SyncStatus } from '../sync/manager.js'; +import { type SyncStatus } from '../sync/git/manager.js'; import { dispatchMemory, isMemoryVerb, type MemoryVerb } from './actions.js'; import { isAllowedHost, isAllowedOrigin, loopbackHosts } from './guards.js'; import { handleMcp } from './mcp-http.js'; diff --git a/src/lib/daemon-client.ts b/src/lib/daemon-client.ts index fa3f943..20cd306 100644 --- a/src/lib/daemon-client.ts +++ b/src/lib/daemon-client.ts @@ -9,9 +9,9 @@ import { type WriteResult, } from '@agentage/memory-core'; import { EADDRINUSE_EXIT_CODE, readDaemonToken, resolvePort } from '../daemon/lifecycle.js'; -import { type SyncResult } from '../sync/cycle.js'; +import { type SyncResult } from '../sync/git/cycle.js'; import { type CouchSyncResult } from '../sync/couch/manager.js'; -import { type SyncStatus } from '../sync/manager.js'; +import { type SyncStatus } from '../sync/git/manager.js'; import { VERSION } from '../utils/version.js'; // One vault syncs on exactly one channel; /api/sync/run yields whichever result fits the vault. diff --git a/src/lib/vault-registry.ts b/src/lib/vault-registry.ts index 4bf17bd..2bafb5e 100644 --- a/src/lib/vault-registry.ts +++ b/src/lib/vault-registry.ts @@ -7,7 +7,7 @@ import { type VaultEntry, type VaultsConfig, } from '@agentage/memory-core'; -import { redactRemoteUrl } from '../sync/remote-url.js'; +import { redactRemoteUrl } from '../sync/git/remote-url.js'; import { isValidVaultName } from './vaults.schema.js'; // Offline registry operations over the unified vaults.json (object map keyed by name). No diff --git a/src/sync/couch/cycle.ts b/src/sync/couch/cycle.ts new file mode 100644 index 0000000..5438e0e --- /dev/null +++ b/src/sync/couch/cycle.ts @@ -0,0 +1,60 @@ +import { type CouchRuntime, type CouchSyncResult, type TargetState } from './manager.types.js'; +import { ensureWire, pendingCount } from './wire.js'; + +// One couch cycle: commit dirty local truth first, drain queued pushes/deletions, then push+pull. +// Every failure is caught and recorded (lastError / paused); it never throws to the caller. +export const runCouchCycle = async ( + rt: CouchRuntime, + st: TargetState +): Promise => { + const vault = st.target.vault; + const build = (extra: Partial): CouchSyncResult => ({ + vault, + channel: 'couch', + ok: true, + committed: false, + pulled: false, + pendingCount: pendingCount(st), + ...extra, + }); + if (st.running) return build({}); + st.running = true; + try { + const bearer = await rt.getBearer(); + if (!bearer) { + st.paused = 'signed out'; + st.lastError = undefined; + return build({ paused: 'signed out' }); + } + const decision = await rt.discovery.channelFor(vault, bearer); + if (decision.kind === 'paused') { + st.paused = decision.reason; + st.lastError = undefined; + return build({ paused: decision.reason }); + } + st.paused = undefined; + const couch = await ensureWire(rt, st, decision); + const pre = await rt.commitDirty(st.target.path, `sync: ${rt.nowIso()}`); + await couch.flushPending(); // drain queued pushes AND queued deletions first + const res = await couch.syncNow(); // pushAll + reconcile deletions, then pullOnce + const post = await rt.commitDirty(st.target.path, `sync: couch ${rt.nowIso()}`); + if (res.error) { + st.lastError = res.error; + return build({ + ok: false, + committed: pre.committed, + pulled: post.committed, + error: res.error, + }); + } + st.lastSync = rt.nowIso(); + st.lastError = undefined; + return build({ committed: pre.committed, pulled: post.committed }); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + st.lastError = msg; + return build({ ok: false, error: msg }); + } finally { + st.running = false; + } +}; diff --git a/src/sync/couch/local-commit.ts b/src/sync/couch/local-commit.ts new file mode 100644 index 0000000..a8d5e9b --- /dev/null +++ b/src/sync/couch/local-commit.ts @@ -0,0 +1,22 @@ +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import { createSyncGit, GitError } from '../git/git-exec.js'; +import { type CommitOutcome } from './manager.types.js'; + +// The default local-git commit: stage everything and make one commit when the tree is dirty. An +// index.lock collision (the engine mid-mutation) is a clean skip - the change stays for next cycle. +export const gitCommitDirty = async (path: string, message: string): Promise => { + if (!existsSync(path)) return { committed: false, skipped: false }; + const git = createSyncGit(path); + try { + if (!existsSync(join(path, '.git'))) await git.run(['init', '-b', 'main']); + await git.run(['add', '-A']); + if ((await git.exec(['diff', '--cached', '--quiet'])).code === 0) + return { committed: false, skipped: false }; + await git.run(['commit', '-m', message]); + return { committed: true, skipped: false }; + } catch (err) { + if (err instanceof GitError && err.kind === 'lock') return { committed: false, skipped: true }; + throw err; + } +}; diff --git a/src/sync/couch/manager.fixtures.ts b/src/sync/couch/manager.fixtures.ts new file mode 100644 index 0000000..6d9a668 --- /dev/null +++ b/src/sync/couch/manager.fixtures.ts @@ -0,0 +1,63 @@ +import { vi } from 'vitest'; +import { type FileStore, type VaultsConfig } from '@agentage/memory-core'; +import { type ChannelDecision, type Discovery } from './discovery.js'; +import { createCouchSyncManager, type CouchSyncManagerDeps } from './manager.js'; + +export const config: VaultsConfig = { + version: 1, + default: 'acct', + vaults: { + acct: { path: '/tmp/acct', origin: [{ remote: 'agentage', interval: 0 }] }, + git: { path: '/tmp/git', origin: [{ remote: 'git@h:g.git' }] }, + local: { path: '/tmp/local' }, + }, +}; + +export const autoConfig: VaultsConfig = { + version: 1, + default: 'acct', + vaults: { + acct: { path: '/tmp/acct', origin: [{ remote: 'agentage', interval: 300 }] }, + two: { path: '/tmp/two', origin: [{ remote: 'agentage', interval: 300 }] }, + }, +}; + +export const noopStore = (): FileStore => ({ + listMarkdown: async () => [], + read: async () => null, + write: async () => {}, + remove: async () => {}, +}); + +export const couchDecision: ChannelDecision = { + kind: 'couch', + endpoint: 'https://couch.test', + db: 'mem_acct', + tokenUrl: 'https://auth.test/couch-token', +}; + +export const makeManager = (over: Partial = {}) => { + const couch = { + pushFileLive: vi.fn(async () => {}), + removeFile: vi.fn(async () => {}), + flushPending: vi.fn(async () => {}), + syncNow: vi.fn(async () => ({ pushed: true, pulled: true })), + }; + const discovery: Discovery = { + channelFor: vi.fn(async () => couchDecision), + reset: vi.fn(), + }; + const mgr = createCouchSyncManager({ + getConfig: () => config, + configDir: () => '/tmp/cfg', + getBearer: async () => 'tok', + discovery, + makeCouchSync: () => couch, + makeFileStore: noopStore, + makeStatePersistence: () => ({ load: async () => null, save: async () => {} }), + commitDirty: async () => ({ committed: false, skipped: false }), + now: () => '2026-01-01T00:00:00Z', + ...over, + }); + return { mgr, couch, discovery }; +}; diff --git a/src/sync/couch/manager.test.ts b/src/sync/couch/manager.test.ts index 91b7154..b3c982b 100644 --- a/src/sync/couch/manager.test.ts +++ b/src/sync/couch/manager.test.ts @@ -1,218 +1,8 @@ import { describe, expect, it, vi } from 'vitest'; -import { type FileStore, type VaultsConfig } from '@agentage/memory-core'; -import { type ChannelDecision, type Discovery } from './discovery.js'; -import { - createCouchSyncManager, - resolveMutationTarget, - type CouchSyncManagerDeps, -} from './manager.js'; - -const config: VaultsConfig = { - version: 1, - default: 'acct', - vaults: { - acct: { path: '/tmp/acct', origin: [{ remote: 'agentage', interval: 0 }] }, - git: { path: '/tmp/git', origin: [{ remote: 'git@h:g.git' }] }, - local: { path: '/tmp/local' }, - }, -}; - -const noopStore = (): FileStore => ({ - listMarkdown: async () => [], - read: async () => null, - write: async () => {}, - remove: async () => {}, -}); - -const couchDecision: ChannelDecision = { - kind: 'couch', - endpoint: 'https://couch.test', - db: 'mem_acct', - tokenUrl: 'https://auth.test/couch-token', -}; - -const makeManager = (over: Partial = {}) => { - const couch = { - pushFileLive: vi.fn(async () => {}), - removeFile: vi.fn(async () => {}), - flushPending: vi.fn(async () => {}), - syncNow: vi.fn(async () => ({ pushed: true, pulled: true })), - }; - const discovery: Discovery = { - channelFor: vi.fn(async () => couchDecision), - reset: vi.fn(), - }; - const mgr = createCouchSyncManager({ - getConfig: () => config, - configDir: () => '/tmp/cfg', - getBearer: async () => 'tok', - discovery, - makeCouchSync: () => couch, - makeFileStore: noopStore, - makeStatePersistence: () => ({ load: async () => null, save: async () => {} }), - commitDirty: async () => ({ committed: false, skipped: false }), - now: () => '2026-01-01T00:00:00Z', - ...over, - }); - return { mgr, couch, discovery }; -}; - -const autoConfig: VaultsConfig = { - version: 1, - default: 'acct', - vaults: { - acct: { path: '/tmp/acct', origin: [{ remote: 'agentage', interval: 300 }] }, - two: { path: '/tmp/two', origin: [{ remote: 'agentage', interval: 300 }] }, - }, -}; - -describe('resolveMutationTarget', () => { - it('maps a bare ref to the default vault when it is an account vault', () => { - expect(resolveMutationTarget(config, { ref: 'notes/x.md' })).toEqual({ - vault: 'acct', - path: 'notes/x.md', - }); - }); - - it('honours an explicit @vault/ prefix', () => { - expect(resolveMutationTarget(config, { ref: '@acct/a/b.md' })).toEqual({ - vault: 'acct', - path: 'a/b.md', - }); - }); - - it('honours opts.vault over the default', () => { - expect(resolveMutationTarget(config, { ref: 'z.md', opts: { vault: 'acct' } })).toEqual({ - vault: 'acct', - path: 'z.md', - }); - }); - - it('returns null for git/local vaults and for non-file refs', () => { - expect(resolveMutationTarget(config, { ref: '@git/z.md' })).toBeNull(); - expect(resolveMutationTarget(config, { ref: 'z.md', opts: { vault: 'local' } })).toBeNull(); - expect(resolveMutationTarget(config, { ref: '@acct' })).toBeNull(); - expect(resolveMutationTarget(config, {})).toBeNull(); - }); - - it('resolves a single-vault config with no default', () => { - const single: VaultsConfig = { - version: 1, - vaults: { only: { path: '/tmp/only', origin: [{ remote: 'agentage' }] } }, - }; - expect(resolveMutationTarget(single, { ref: 'n.md' })).toEqual({ vault: 'only', path: 'n.md' }); - }); -}); - -describe('createCouchSyncManager.onWrite', () => { - it('pushes a write on an account vault via the couch channel', async () => { - const { mgr, couch } = makeManager(); - mgr.onWrite('write', { ref: 'notes/x.md' }); - await vi.waitFor(() => expect(couch.pushFileLive).toHaveBeenCalledWith('notes/x.md')); - expect(couch.removeFile).not.toHaveBeenCalled(); - }); - - it('tombstones a delete on an account vault', async () => { - const { mgr, couch } = makeManager(); - mgr.onWrite('delete', { ref: 'notes/x.md' }); - await vi.waitFor(() => expect(couch.removeFile).toHaveBeenCalledWith('notes/x.md')); - expect(couch.pushFileLive).not.toHaveBeenCalled(); - }); - - it('never touches couch for a git or local vault mutation', async () => { - const { mgr, couch, discovery } = makeManager(); - mgr.onWrite('write', { ref: '@git/z.md' }); - mgr.onWrite('edit', { ref: 'z.md', opts: { vault: 'local' } }); - await new Promise((r) => setTimeout(r, 20)); - expect(couch.pushFileLive).not.toHaveBeenCalled(); - expect(discovery.channelFor).not.toHaveBeenCalled(); - }); - - it('does not fire for read/search/list verbs', async () => { - const { mgr, couch } = makeManager(); - mgr.onWrite('read', { ref: 'notes/x.md' }); - mgr.onWrite('list', {}); - await new Promise((r) => setTimeout(r, 20)); - expect(couch.pushFileLive).not.toHaveBeenCalled(); - }); - - it('enqueues (no network) when signed out', async () => { - const enqueued: string[] = []; - const { mgr, couch, discovery } = makeManager({ - getBearer: async () => null, - makeStatePersistence: () => ({ - load: async () => null, - save: async (s) => { - enqueued.push(...s.pending); - }, - }), - }); - mgr.onWrite('write', { ref: 'notes/x.md' }); - await vi.waitFor(() => expect(enqueued).toContain('notes/x.md')); - expect(discovery.channelFor).not.toHaveBeenCalled(); - expect(couch.pushFileLive).not.toHaveBeenCalled(); - }); - - it('a signed-out delete enqueues a durable deletion, zero network', async () => { - const deletions: string[] = []; - const { mgr, couch, discovery } = makeManager({ - getBearer: async () => null, - makeStatePersistence: () => ({ - load: async () => null, - save: async (s) => { - deletions.push(...(s.deletions ?? [])); - }, - }), - }); - mgr.onWrite('delete', { ref: 'notes/x.md' }); - await vi.waitFor(() => expect(deletions).toContain('notes/x.md')); - expect(discovery.channelFor).not.toHaveBeenCalled(); - expect(couch.removeFile).not.toHaveBeenCalled(); - }); - - it('defers a write to the pending queue when signed in but the channel is not couch-ready', async () => { - const pending: string[] = []; - const { mgr, couch } = makeManager({ - discovery: { - channelFor: vi.fn(async (): Promise => ({ - kind: 'paused', - reason: 'not provisioned', - })), - reset: vi.fn(), - }, - makeStatePersistence: () => ({ - load: async () => null, - save: async (s) => { - pending.push(...s.pending); - }, - }), - }); - mgr.onWrite('write', { ref: 'notes/y.md' }); - await vi.waitFor(() => expect(pending).toContain('notes/y.md')); - expect(couch.pushFileLive).not.toHaveBeenCalled(); - }); - - it('a delete surviving a discovery failure is enqueued, never dropped', async () => { - const deletions: string[] = []; - const { mgr, couch } = makeManager({ - discovery: { - channelFor: vi.fn(async () => { - throw new Error('well-known unreachable'); - }), - reset: vi.fn(), - }, - makeStatePersistence: () => ({ - load: async () => null, - save: async (s) => { - deletions.push(...(s.deletions ?? [])); - }, - }), - }); - mgr.onWrite('delete', { ref: 'notes/x.md' }); - await vi.waitFor(() => expect(deletions).toContain('notes/x.md')); - expect(couch.removeFile).not.toHaveBeenCalled(); - }); -}); +import { type VaultsConfig } from '@agentage/memory-core'; +import { type ChannelDecision } from './discovery.js'; +import { createCouchSyncManager } from './manager.js'; +import { autoConfig, config, couchDecision, makeManager, noopStore } from './manager.fixtures.js'; describe('createCouchSyncManager.status and runNow', () => { it('reports one couch target with its cadence and zero pending', () => { diff --git a/src/sync/couch/manager.ts b/src/sync/couch/manager.ts index f7a122c..13e919c 100644 --- a/src/sync/couch/manager.ts +++ b/src/sync/couch/manager.ts @@ -1,158 +1,35 @@ -import { existsSync } from 'node:fs'; -import { join } from 'node:path'; -import { - CouchSync, - CouchTokenClient, - createCouchState, - isAccountVault, - type CouchState, - type CouchStatePersistence, - type FetchLike, - type FetchJson, - type FileStore, - type SyncResult as CouchChannelResult, - type VaultsConfig, -} from '@agentage/memory-core'; +import { CouchSync, type FetchLike, type FetchJson } from '@agentage/memory-core'; import { currentBearer } from '../../lib/api.js'; import { getConfigDir, readAuth } from '../../lib/config.js'; import { links, siteFqdn } from '../../lib/origins.js'; import { defaultProvisionDeps, provisionAccountVault } from '../../lib/provision.js'; import { loadVaultsConfig } from '../../lib/vaults.js'; -import { type MemoryVerb } from '../../daemon/actions.js'; -import { createSyncGit, GitError } from '../git-exec.js'; -import { intervalMs } from '../planner.js'; -import { createDiscovery, type ChannelDecision, type Discovery } from './discovery.js'; +import { intervalMs } from '../git/planner.js'; +import { runCouchCycle } from './cycle.js'; +import { createDiscovery } from './discovery.js'; import { createFileStore } from './file-store.js'; +import { gitCommitDirty } from './local-commit.js'; +import { + type CouchRuntime, + type CouchSyncManager, + type CouchSyncManagerDeps, + type CouchTargetStatus, + type MakeCouchSync, + type TargetState, +} from './manager.types.js'; +import { resolveMutationTarget } from './mutation-target.js'; +import { pushOnWrite } from './push-on-write.js'; import { createStatePersistence } from './state-store.js'; import { autoCouchTargets, couchTargets, type CouchTarget } from './targets.js'; +import { getState, pendingCount } from './wire.js'; -export interface CouchSyncResult { - vault: string; - channel: 'couch'; - ok: boolean; - committed: boolean; // committed local dirty changes before the push - pulled: boolean; // a pull applied changes and they were committed - pendingCount: number; - paused?: string; // set when the target is paused (signed out / not provisioned) - error?: string; -} - -export interface CouchTargetStatus { - vault: string; - channel: 'couch'; - intervalSeconds: number; - lastSync?: string; - lastError?: string; - pendingCount: number; - paused?: string; - running: boolean; -} - -// What the manager uses from a CouchSync - the real class satisfies it; tests inject a mock. -interface CouchLike { - pushFileLive(path: string): Promise; - removeFile(path: string): Promise; - flushPending(): Promise; - syncNow(): Promise; -} - -type MakeCouchSync = ( - files: FileStore, - cfg: { endpoint: string; db: string }, - fetch: FetchLike, - authorize: () => Promise, - onUnauthorized: () => void, - state: CouchState, - log?: (msg: string) => void -) => CouchLike; - -interface CommitOutcome { - committed: boolean; - skipped: boolean; // an index.lock collision - retried next cycle -} - -export interface CouchSyncManagerDeps { - getConfig?: () => VaultsConfig; - configDir?: () => string; - getBearer?: () => Promise; - discovery?: Discovery; - fetch?: FetchLike; - makeFileStore?: (path: string) => FileStore; - makeStatePersistence?: (configDir: string, vault: string) => CouchStatePersistence; - makeCouchSync?: MakeCouchSync; - commitDirty?: (path: string, message: string) => Promise; - now?: () => string; // ISO timestamp - log?: (msg: string) => void; -} - -export interface CouchSyncManager { - reschedule(): void; - runNow(vault: string): Promise; - onWrite(verb: MemoryVerb, body: unknown): void; - status(): CouchTargetStatus[]; - stop(): void; -} - -interface TargetState { - target: CouchTarget; - files: FileStore; - state?: CouchState; - statePromise?: Promise; - couch?: CouchLike; - wireKey?: string; - running: boolean; - lastSync?: string; - lastError?: string; - paused?: string; -} - -// Map one memory-verb wire payload to the account vault + vault-relative POSIX path it mutated, or -// null when the target is not an account vault (git/local mutations never touch the couch channel). -export const resolveMutationTarget = ( - config: VaultsConfig, - body: unknown -): { vault: string; path: string } | null => { - const p = (body ?? {}) as { ref?: unknown; opts?: { vault?: unknown } }; - const ref = typeof p.ref === 'string' ? p.ref : ''; - if (!ref) return null; - let vault: string | undefined; - let path: string; - if (ref.startsWith('@')) { - const m = ref.match(/^@([^/]+)\/(.+)$/); - if (!m) return null; // a bare '@vault' is not a file mutation - vault = m[1]; - path = m[2] as string; - } else { - vault = (typeof p.opts?.vault === 'string' ? p.opts.vault : undefined) ?? config.default; - if (!vault) { - const names = Object.keys(config.vaults ?? {}); - if (names.length === 1) vault = names[0]; - } - path = ref; - } - if (!vault) return null; - const entry = config.vaults?.[vault]; - if (!entry || !isAccountVault(entry)) return null; - return { vault, path: path.replace(/^\.?\//, '') }; -}; - -// The default local-git commit: stage everything and make one commit when the tree is dirty. An -// index.lock collision (the engine mid-mutation) is a clean skip - the change stays for next cycle. -const gitCommitDirty = async (path: string, message: string): Promise => { - if (!existsSync(path)) return { committed: false, skipped: false }; - const git = createSyncGit(path); - try { - if (!existsSync(join(path, '.git'))) await git.run(['init', '-b', 'main']); - await git.run(['add', '-A']); - if ((await git.exec(['diff', '--cached', '--quiet'])).code === 0) - return { committed: false, skipped: false }; - await git.run(['commit', '-m', message]); - return { committed: true, skipped: false }; - } catch (err) { - if (err instanceof GitError && err.kind === 'lock') return { committed: false, skipped: true }; - throw err; - } -}; +export { + type CouchSyncManager, + type CouchSyncManagerDeps, + type CouchSyncResult, + type CouchTargetStatus, +} from './manager.types.js'; +export { resolveMutationTarget } from './mutation-target.js'; const defaultFetch: FetchLike = (url, init) => globalThis.fetch(url, init as RequestInit); @@ -167,33 +44,33 @@ const defaultFetchJson: FetchJson = async (url, token) => { // the daemon and never blocks a memory API response. export const createCouchSyncManager = (deps: CouchSyncManagerDeps = {}): CouchSyncManager => { const getConfig = deps.getConfig ?? (() => loadVaultsConfig().config); - const configDir = deps.configDir ?? getConfigDir; const getBearer = deps.getBearer ?? (() => currentBearer(readAuth, links(siteFqdn()))); - const fetch = deps.fetch ?? defaultFetch; const makeFileStore = deps.makeFileStore ?? createFileStore; - const makeStatePersistence = deps.makeStatePersistence ?? createStatePersistence; const makeCouchSync: MakeCouchSync = deps.makeCouchSync ?? ((files, cfg, f, authorize, onUnauthorized, state, log) => new CouchSync(files, cfg, f, authorize, onUnauthorized, state, log)); - const commitDirty = deps.commitDirty ?? gitCommitDirty; - const nowIso = deps.now ?? (() => new Date().toISOString()); - const log = deps.log ?? (() => {}); - const discovery = - deps.discovery ?? - createDiscovery({ - bootstrapHost: links(siteFqdn()).sync, - fetchJson: defaultFetchJson, - provision: (vault) => provisionAccountVault(vault, defaultProvisionDeps()), - }); + const rt: CouchRuntime = { + configDir: deps.configDir ?? getConfigDir, + getBearer, + fetch: deps.fetch ?? defaultFetch, + makeCouchSync, + makeStatePersistence: deps.makeStatePersistence ?? createStatePersistence, + commitDirty: deps.commitDirty ?? gitCommitDirty, + discovery: + deps.discovery ?? + createDiscovery({ + bootstrapHost: links(siteFqdn()).sync, + fetchJson: defaultFetchJson, + provision: (vault) => provisionAccountVault(vault, defaultProvisionDeps()), + }), + nowIso: deps.now ?? (() => new Date().toISOString()), + log: deps.log ?? (() => {}), + }; const states = new Map(); const timers = new Map(); - // Everything queued for retry: failed/deferred pushes plus not-yet-tombstoned deletions. - const pendingCount = (st: TargetState | undefined): number => - st?.state ? st.state.pendingPaths().length + st.state.deletionPaths().length : 0; - const ensureTargetState = (target: CouchTarget): TargetState => { const existing = states.get(target.vault); if (existing) { @@ -205,111 +82,6 @@ export const createCouchSyncManager = (deps: CouchSyncManagerDeps = {}): CouchSy return fresh; }; - const getState = (st: TargetState): Promise => - (st.statePromise ??= createCouchState(makeStatePersistence(configDir(), st.target.vault)).then( - (s) => (st.state = s) - )); - - const ensureWire = async (st: TargetState, d: ChannelDecision): Promise => { - if (d.kind !== 'couch') throw new Error('ensureWire: not a couch channel'); - const key = `${d.endpoint}|${d.db}|${d.tokenUrl}`; - if (st.couch && st.wireKey === key) return st.couch; - const state = await getState(st); - const tokens = new CouchTokenClient(d.tokenUrl, st.target.vault, fetch, getBearer, Date.now); - st.couch = makeCouchSync( - st.files, - { endpoint: d.endpoint, db: d.db }, - fetch, - () => tokens.token(), - () => tokens.invalidate(), - state, - log - ); - st.wireKey = key; - return st.couch; - }; - - const cycle = async (st: TargetState): Promise => { - const vault = st.target.vault; - const build = (extra: Partial): CouchSyncResult => ({ - vault, - channel: 'couch', - ok: true, - committed: false, - pulled: false, - pendingCount: pendingCount(st), - ...extra, - }); - if (st.running) return build({}); - st.running = true; - try { - const bearer = await getBearer(); - if (!bearer) { - st.paused = 'signed out'; - st.lastError = undefined; - return build({ paused: 'signed out' }); - } - const decision = await discovery.channelFor(vault, bearer); - if (decision.kind === 'paused') { - st.paused = decision.reason; - st.lastError = undefined; - return build({ paused: decision.reason }); - } - st.paused = undefined; - const couch = await ensureWire(st, decision); - const pre = await commitDirty(st.target.path, `sync: ${nowIso()}`); - await couch.flushPending(); // drain queued pushes AND queued deletions first - const res = await couch.syncNow(); // pushAll + reconcile deletions, then pullOnce - const post = await commitDirty(st.target.path, `sync: couch ${nowIso()}`); - if (res.error) { - st.lastError = res.error; - return build({ - ok: false, - committed: pre.committed, - pulled: post.committed, - error: res.error, - }); - } - st.lastSync = nowIso(); - st.lastError = undefined; - return build({ committed: pre.committed, pulled: post.committed }); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - st.lastError = msg; - return build({ ok: false, error: msg }); - } finally { - st.running = false; - } - }; - - // Sync-on-save: push (or tombstone) one path right after the engine committed it. Failures queue - // the path in the module's persisted pending/deletion sets (retried by the next cycle) and never - // surface to the API. A delete is durable regardless of auth/network state at delete time: with - // no wire it enqueues the deletion, and removeFile itself self-enqueues on transport failure. - const pushOnWrite = async (st: TargetState, verb: MemoryVerb, path: string): Promise => { - const defer = async (): Promise => { - const state = await getState(st); - if (verb === 'delete') await state.enqueueDeletion(path); - else await state.enqueue(path); - }; - try { - const bearer = await getBearer(); - if (bearer) { - const decision = await discovery.channelFor(st.target.vault, bearer); - if (decision.kind === 'couch') { - const couch = await ensureWire(st, decision); - if (verb === 'delete') await couch.removeFile(path); - else await couch.pushFileLive(path); - return; - } - } - await defer(); // no wire yet (signed out / paused) - queued until one exists - } catch (err) { - log(`couch push-on-write ${path}: ${err instanceof Error ? err.message : String(err)}`); - await defer().catch(() => {}); - } - }; - return { reschedule() { for (const timer of timers.values()) clearInterval(timer); @@ -317,10 +89,10 @@ export const createCouchSyncManager = (deps: CouchSyncManagerDeps = {}): CouchSy const targets = couchTargets(getConfig()); const live = new Set(targets.map((t) => t.vault)); for (const vault of [...states.keys()]) if (!live.has(vault)) states.delete(vault); - for (const t of targets) void getState(ensureTargetState(t)).catch(() => {}); + for (const t of targets) void getState(rt, ensureTargetState(t)).catch(() => {}); for (const t of autoCouchTargets(getConfig())) { const timer = setInterval( - () => void cycle(ensureTargetState(t)), + () => void runCouchCycle(rt, ensureTargetState(t)), intervalMs(t.intervalSeconds) ); timer.unref?.(); @@ -330,7 +102,7 @@ export const createCouchSyncManager = (deps: CouchSyncManagerDeps = {}): CouchSy async runNow(vault) { const t = couchTargets(getConfig()).find((x) => x.vault === vault); if (!t) throw new Error(`'${vault}' is not an account vault`); - return cycle(ensureTargetState(t)); + return runCouchCycle(rt, ensureTargetState(t)); }, onWrite(verb, body) { if (verb !== 'write' && verb !== 'edit' && verb !== 'delete') return; @@ -338,7 +110,7 @@ export const createCouchSyncManager = (deps: CouchSyncManagerDeps = {}): CouchSy if (!target) return; const t = couchTargets(getConfig()).find((x) => x.vault === target.vault); if (!t) return; - void pushOnWrite(ensureTargetState(t), verb, target.path); + void pushOnWrite(rt, ensureTargetState(t), verb, target.path); }, status() { return couchTargets(getConfig()).map((t): CouchTargetStatus => { diff --git a/src/sync/couch/manager.types.ts b/src/sync/couch/manager.types.ts new file mode 100644 index 0000000..271d189 --- /dev/null +++ b/src/sync/couch/manager.types.ts @@ -0,0 +1,104 @@ +import { + type CouchState, + type CouchStatePersistence, + type FetchLike, + type FileStore, + type SyncResult as CouchChannelResult, + type VaultsConfig, +} from '@agentage/memory-core'; +import { type MemoryVerb } from '../../daemon/actions.js'; +import { type Discovery } from './discovery.js'; +import { type CouchTarget } from './targets.js'; + +export interface CouchSyncResult { + vault: string; + channel: 'couch'; + ok: boolean; + committed: boolean; // committed local dirty changes before the push + pulled: boolean; // a pull applied changes and they were committed + pendingCount: number; + paused?: string; // set when the target is paused (signed out / not provisioned) + error?: string; +} + +export interface CouchTargetStatus { + vault: string; + channel: 'couch'; + intervalSeconds: number; + lastSync?: string; + lastError?: string; + pendingCount: number; + paused?: string; + running: boolean; +} + +// What the manager uses from a CouchSync - the real class satisfies it; tests inject a mock. +export interface CouchLike { + pushFileLive(path: string): Promise; + removeFile(path: string): Promise; + flushPending(): Promise; + syncNow(): Promise; +} + +export type MakeCouchSync = ( + files: FileStore, + cfg: { endpoint: string; db: string }, + fetch: FetchLike, + authorize: () => Promise, + onUnauthorized: () => void, + state: CouchState, + log?: (msg: string) => void +) => CouchLike; + +export interface CommitOutcome { + committed: boolean; + skipped: boolean; // an index.lock collision - retried next cycle +} + +export interface CouchSyncManagerDeps { + getConfig?: () => VaultsConfig; + configDir?: () => string; + getBearer?: () => Promise; + discovery?: Discovery; + fetch?: FetchLike; + makeFileStore?: (path: string) => FileStore; + makeStatePersistence?: (configDir: string, vault: string) => CouchStatePersistence; + makeCouchSync?: MakeCouchSync; + commitDirty?: (path: string, message: string) => Promise; + now?: () => string; // ISO timestamp + log?: (msg: string) => void; +} + +export interface CouchSyncManager { + reschedule(): void; + runNow(vault: string): Promise; + onWrite(verb: MemoryVerb, body: unknown): void; + status(): CouchTargetStatus[]; + stop(): void; +} + +export interface TargetState { + target: CouchTarget; + files: FileStore; + state?: CouchState; + statePromise?: Promise; + couch?: CouchLike; + wireKey?: string; + running: boolean; + lastSync?: string; + lastError?: string; + paused?: string; +} + +// Resolved deps shared by the extracted wire/cycle/push-on-write seams; built once by the manager. +export interface CouchRuntime { + configDir: () => string; + getBearer: () => Promise; + fetch: FetchLike; + makeCouchSync: MakeCouchSync; + makeStatePersistence: (configDir: string, vault: string) => CouchStatePersistence; + commitDirty: (path: string, message: string) => Promise; + discovery: Discovery; + nowIso: () => string; + log: (msg: string) => void; +} diff --git a/src/sync/couch/mutation-target.test.ts b/src/sync/couch/mutation-target.test.ts new file mode 100644 index 0000000..9af0db7 --- /dev/null +++ b/src/sync/couch/mutation-target.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, it } from 'vitest'; +import { type VaultsConfig } from '@agentage/memory-core'; +import { resolveMutationTarget } from './mutation-target.js'; + +const config: VaultsConfig = { + version: 1, + default: 'acct', + vaults: { + acct: { path: '/tmp/acct', origin: [{ remote: 'agentage', interval: 0 }] }, + git: { path: '/tmp/git', origin: [{ remote: 'git@h:g.git' }] }, + local: { path: '/tmp/local' }, + }, +}; + +describe('resolveMutationTarget', () => { + it('maps a bare ref to the default vault when it is an account vault', () => { + expect(resolveMutationTarget(config, { ref: 'notes/x.md' })).toEqual({ + vault: 'acct', + path: 'notes/x.md', + }); + }); + + it('honours an explicit @vault/ prefix', () => { + expect(resolveMutationTarget(config, { ref: '@acct/a/b.md' })).toEqual({ + vault: 'acct', + path: 'a/b.md', + }); + }); + + it('honours opts.vault over the default', () => { + expect(resolveMutationTarget(config, { ref: 'z.md', opts: { vault: 'acct' } })).toEqual({ + vault: 'acct', + path: 'z.md', + }); + }); + + it('returns null for git/local vaults and for non-file refs', () => { + expect(resolveMutationTarget(config, { ref: '@git/z.md' })).toBeNull(); + expect(resolveMutationTarget(config, { ref: 'z.md', opts: { vault: 'local' } })).toBeNull(); + expect(resolveMutationTarget(config, { ref: '@acct' })).toBeNull(); + expect(resolveMutationTarget(config, {})).toBeNull(); + }); + + it('resolves a single-vault config with no default', () => { + const single: VaultsConfig = { + version: 1, + vaults: { only: { path: '/tmp/only', origin: [{ remote: 'agentage' }] } }, + }; + expect(resolveMutationTarget(single, { ref: 'n.md' })).toEqual({ vault: 'only', path: 'n.md' }); + }); +}); diff --git a/src/sync/couch/mutation-target.ts b/src/sync/couch/mutation-target.ts new file mode 100644 index 0000000..c31eec4 --- /dev/null +++ b/src/sync/couch/mutation-target.ts @@ -0,0 +1,31 @@ +import { isAccountVault, type VaultsConfig } from '@agentage/memory-core'; + +// Map one memory-verb wire payload to the account vault + vault-relative POSIX path it mutated, or +// null when the target is not an account vault (git/local mutations never touch the couch channel). +export const resolveMutationTarget = ( + config: VaultsConfig, + body: unknown +): { vault: string; path: string } | null => { + const p = (body ?? {}) as { ref?: unknown; opts?: { vault?: unknown } }; + const ref = typeof p.ref === 'string' ? p.ref : ''; + if (!ref) return null; + let vault: string | undefined; + let path: string; + if (ref.startsWith('@')) { + const m = ref.match(/^@([^/]+)\/(.+)$/); + if (!m) return null; // a bare '@vault' is not a file mutation + vault = m[1]; + path = m[2] as string; + } else { + vault = (typeof p.opts?.vault === 'string' ? p.opts.vault : undefined) ?? config.default; + if (!vault) { + const names = Object.keys(config.vaults ?? {}); + if (names.length === 1) vault = names[0]; + } + path = ref; + } + if (!vault) return null; + const entry = config.vaults?.[vault]; + if (!entry || !isAccountVault(entry)) return null; + return { vault, path: path.replace(/^\.?\//, '') }; +}; diff --git a/src/sync/couch/push-on-write.test.ts b/src/sync/couch/push-on-write.test.ts new file mode 100644 index 0000000..ad399bc --- /dev/null +++ b/src/sync/couch/push-on-write.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it, vi } from 'vitest'; +import { type ChannelDecision } from './discovery.js'; +import { makeManager } from './manager.fixtures.js'; + +describe('createCouchSyncManager.onWrite', () => { + it('pushes a write on an account vault via the couch channel', async () => { + const { mgr, couch } = makeManager(); + mgr.onWrite('write', { ref: 'notes/x.md' }); + await vi.waitFor(() => expect(couch.pushFileLive).toHaveBeenCalledWith('notes/x.md')); + expect(couch.removeFile).not.toHaveBeenCalled(); + }); + + it('tombstones a delete on an account vault', async () => { + const { mgr, couch } = makeManager(); + mgr.onWrite('delete', { ref: 'notes/x.md' }); + await vi.waitFor(() => expect(couch.removeFile).toHaveBeenCalledWith('notes/x.md')); + expect(couch.pushFileLive).not.toHaveBeenCalled(); + }); + + it('never touches couch for a git or local vault mutation', async () => { + const { mgr, couch, discovery } = makeManager(); + mgr.onWrite('write', { ref: '@git/z.md' }); + mgr.onWrite('edit', { ref: 'z.md', opts: { vault: 'local' } }); + await new Promise((r) => setTimeout(r, 20)); + expect(couch.pushFileLive).not.toHaveBeenCalled(); + expect(discovery.channelFor).not.toHaveBeenCalled(); + }); + + it('does not fire for read/search/list verbs', async () => { + const { mgr, couch } = makeManager(); + mgr.onWrite('read', { ref: 'notes/x.md' }); + mgr.onWrite('list', {}); + await new Promise((r) => setTimeout(r, 20)); + expect(couch.pushFileLive).not.toHaveBeenCalled(); + }); + + it('enqueues (no network) when signed out', async () => { + const enqueued: string[] = []; + const { mgr, couch, discovery } = makeManager({ + getBearer: async () => null, + makeStatePersistence: () => ({ + load: async () => null, + save: async (s) => { + enqueued.push(...s.pending); + }, + }), + }); + mgr.onWrite('write', { ref: 'notes/x.md' }); + await vi.waitFor(() => expect(enqueued).toContain('notes/x.md')); + expect(discovery.channelFor).not.toHaveBeenCalled(); + expect(couch.pushFileLive).not.toHaveBeenCalled(); + }); + + it('a signed-out delete enqueues a durable deletion, zero network', async () => { + const deletions: string[] = []; + const { mgr, couch, discovery } = makeManager({ + getBearer: async () => null, + makeStatePersistence: () => ({ + load: async () => null, + save: async (s) => { + deletions.push(...(s.deletions ?? [])); + }, + }), + }); + mgr.onWrite('delete', { ref: 'notes/x.md' }); + await vi.waitFor(() => expect(deletions).toContain('notes/x.md')); + expect(discovery.channelFor).not.toHaveBeenCalled(); + expect(couch.removeFile).not.toHaveBeenCalled(); + }); + + it('defers a write to the pending queue when signed in but the channel is not couch-ready', async () => { + const pending: string[] = []; + const { mgr, couch } = makeManager({ + discovery: { + channelFor: vi.fn(async (): Promise => ({ + kind: 'paused', + reason: 'not provisioned', + })), + reset: vi.fn(), + }, + makeStatePersistence: () => ({ + load: async () => null, + save: async (s) => { + pending.push(...s.pending); + }, + }), + }); + mgr.onWrite('write', { ref: 'notes/y.md' }); + await vi.waitFor(() => expect(pending).toContain('notes/y.md')); + expect(couch.pushFileLive).not.toHaveBeenCalled(); + }); + + it('a delete surviving a discovery failure is enqueued, never dropped', async () => { + const deletions: string[] = []; + const { mgr, couch } = makeManager({ + discovery: { + channelFor: vi.fn(async () => { + throw new Error('well-known unreachable'); + }), + reset: vi.fn(), + }, + makeStatePersistence: () => ({ + load: async () => null, + save: async (s) => { + deletions.push(...(s.deletions ?? [])); + }, + }), + }); + mgr.onWrite('delete', { ref: 'notes/x.md' }); + await vi.waitFor(() => expect(deletions).toContain('notes/x.md')); + expect(couch.removeFile).not.toHaveBeenCalled(); + }); +}); diff --git a/src/sync/couch/push-on-write.ts b/src/sync/couch/push-on-write.ts new file mode 100644 index 0000000..820dd29 --- /dev/null +++ b/src/sync/couch/push-on-write.ts @@ -0,0 +1,36 @@ +import { type MemoryVerb } from '../../daemon/actions.js'; +import { type CouchRuntime, type TargetState } from './manager.types.js'; +import { ensureWire, getState } from './wire.js'; + +// Sync-on-save: push (or tombstone) one path right after the engine committed it. Failures queue +// the path in the module's persisted pending/deletion sets (retried by the next cycle) and never +// surface to the API. A delete is durable regardless of auth/network state at delete time: with +// no wire it enqueues the deletion, and removeFile itself self-enqueues on transport failure. +export const pushOnWrite = async ( + rt: CouchRuntime, + st: TargetState, + verb: MemoryVerb, + path: string +): Promise => { + const defer = async (): Promise => { + const state = await getState(rt, st); + if (verb === 'delete') await state.enqueueDeletion(path); + else await state.enqueue(path); + }; + try { + const bearer = await rt.getBearer(); + if (bearer) { + const decision = await rt.discovery.channelFor(st.target.vault, bearer); + if (decision.kind === 'couch') { + const couch = await ensureWire(rt, st, decision); + if (verb === 'delete') await couch.removeFile(path); + else await couch.pushFileLive(path); + return; + } + } + await defer(); // no wire yet (signed out / paused) - queued until one exists + } catch (err) { + rt.log(`couch push-on-write ${path}: ${err instanceof Error ? err.message : String(err)}`); + await defer().catch(() => {}); + } +}; diff --git a/src/sync/couch/targets.ts b/src/sync/couch/targets.ts index caf36d4..1cc1cf3 100644 --- a/src/sync/couch/targets.ts +++ b/src/sync/couch/targets.ts @@ -1,5 +1,5 @@ import { expandPath, isAccountVault, type VaultsConfig } from '@agentage/memory-core'; -import { DEFAULT_INTERVAL_SECONDS } from '../planner.js'; +import { DEFAULT_INTERVAL_SECONDS } from '../git/planner.js'; // The account (agentage) channel a couch target syncs to. Unlike a git target it has no external // remote URL: the daemon resolves the per-memory CouchDB + JWT endpoints from discovery at runtime. diff --git a/src/sync/couch/wire.ts b/src/sync/couch/wire.ts new file mode 100644 index 0000000..926fd49 --- /dev/null +++ b/src/sync/couch/wire.ts @@ -0,0 +1,41 @@ +import { createCouchState, CouchTokenClient, type CouchState } from '@agentage/memory-core'; +import { type ChannelDecision } from './discovery.js'; +import { type CouchLike, type CouchRuntime, type TargetState } from './manager.types.js'; + +// Everything queued for retry: failed/deferred pushes plus not-yet-tombstoned deletions. +export const pendingCount = (st: TargetState | undefined): number => + st?.state ? st.state.pendingPaths().length + st.state.deletionPaths().length : 0; + +export const getState = (rt: CouchRuntime, st: TargetState): Promise => + (st.statePromise ??= createCouchState( + rt.makeStatePersistence(rt.configDir(), st.target.vault) + ).then((s) => (st.state = s))); + +export const ensureWire = async ( + rt: CouchRuntime, + st: TargetState, + d: ChannelDecision +): Promise => { + if (d.kind !== 'couch') throw new Error('ensureWire: not a couch channel'); + const key = `${d.endpoint}|${d.db}|${d.tokenUrl}`; + if (st.couch && st.wireKey === key) return st.couch; + const state = await getState(rt, st); + const tokens = new CouchTokenClient( + d.tokenUrl, + st.target.vault, + rt.fetch, + rt.getBearer, + Date.now + ); + st.couch = rt.makeCouchSync( + st.files, + { endpoint: d.endpoint, db: d.db }, + rt.fetch, + () => tokens.token(), + () => tokens.invalidate(), + state, + rt.log + ); + st.wireKey = key; + return st.couch; +}; diff --git a/src/sync/conflict.test.ts b/src/sync/git/conflict.test.ts similarity index 100% rename from src/sync/conflict.test.ts rename to src/sync/git/conflict.test.ts diff --git a/src/sync/conflict.ts b/src/sync/git/conflict.ts similarity index 100% rename from src/sync/conflict.ts rename to src/sync/git/conflict.ts diff --git a/src/sync/cycle.test.ts b/src/sync/git/cycle.test.ts similarity index 100% rename from src/sync/cycle.test.ts rename to src/sync/git/cycle.test.ts diff --git a/src/sync/cycle.ts b/src/sync/git/cycle.ts similarity index 100% rename from src/sync/cycle.ts rename to src/sync/git/cycle.ts diff --git a/src/sync/git-exec.test.ts b/src/sync/git/git-exec.test.ts similarity index 100% rename from src/sync/git-exec.test.ts rename to src/sync/git/git-exec.test.ts diff --git a/src/sync/git-exec.ts b/src/sync/git/git-exec.ts similarity index 100% rename from src/sync/git-exec.ts rename to src/sync/git/git-exec.ts diff --git a/src/sync/manager.test.ts b/src/sync/git/manager.test.ts similarity index 100% rename from src/sync/manager.test.ts rename to src/sync/git/manager.test.ts diff --git a/src/sync/manager.ts b/src/sync/git/manager.ts similarity index 96% rename from src/sync/manager.ts rename to src/sync/git/manager.ts index 10b1ec9..7a68f99 100644 --- a/src/sync/manager.ts +++ b/src/sync/git/manager.ts @@ -1,7 +1,7 @@ import { type VaultsConfig } from '@agentage/memory-core'; -import { loadVaultsConfig } from '../lib/vaults.js'; -import { type CouchTargetStatus } from './couch/manager.js'; -import { type DiscoverStatus } from './discover/watcher.js'; +import { loadVaultsConfig } from '../../lib/vaults.js'; +import { type CouchTargetStatus } from '../couch/manager.js'; +import { type DiscoverStatus } from '../discover/watcher.js'; import { runSyncCycle, type SyncResult } from './cycle.js'; import { autoSyncTargets, intervalMs, syncTargets, type SyncTarget } from './planner.js'; diff --git a/src/sync/planner.test.ts b/src/sync/git/planner.test.ts similarity index 100% rename from src/sync/planner.test.ts rename to src/sync/git/planner.test.ts diff --git a/src/sync/planner.ts b/src/sync/git/planner.ts similarity index 100% rename from src/sync/planner.ts rename to src/sync/git/planner.ts diff --git a/src/sync/remote-url.test.ts b/src/sync/git/remote-url.test.ts similarity index 100% rename from src/sync/remote-url.test.ts rename to src/sync/git/remote-url.test.ts diff --git a/src/sync/remote-url.ts b/src/sync/git/remote-url.ts similarity index 100% rename from src/sync/remote-url.ts rename to src/sync/git/remote-url.ts diff --git a/vitest.config.ts b/vitest.config.ts index 2651933..cac61bf 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -6,7 +6,13 @@ export default defineConfig({ coverage: { provider: 'v8', include: ['src/**/*.ts'], - exclude: ['**/*.test.ts', '**/index.ts', 'src/cli.ts', 'src/daemon-entry.ts'], + exclude: [ + '**/*.test.ts', + '**/*.fixtures.ts', + '**/index.ts', + 'src/cli.ts', + 'src/daemon-entry.ts', + ], thresholds: { branches: 65, functions: 70,