From fc967034301398dcfe3a59559ad91ac265cc5dd8 Mon Sep 17 00:00:00 2001 From: Volodymyr Vreshch Date: Mon, 6 Jul 2026 09:50:45 +0200 Subject: [PATCH] feat(sync): git sync engine in the daemon - origin push/pull (M4) --- CLAUDE.md | 8 +- e2e/sync.test.ts | 170 ++++++++++++++++++++++++++++++++ package-lock.json | 12 +-- package.json | 2 +- src/commands/daemon-cmd.ts | 18 +++- src/commands/vault-sync.test.ts | 78 +++++++++++++++ src/commands/vault-sync.ts | 72 ++++++++++++++ src/commands/vault.test.ts | 12 ++- src/commands/vault.ts | 20 +++- src/daemon-entry.ts | 14 ++- src/daemon/server.ts | 23 +++++ src/lib/daemon-client.ts | 29 ++++++ src/sync/conflict.test.ts | 27 +++++ src/sync/conflict.ts | 14 +++ src/sync/cycle.test.ts | 158 +++++++++++++++++++++++++++++ src/sync/cycle.ts | 168 +++++++++++++++++++++++++++++++ src/sync/git-exec.test.ts | 55 +++++++++++ src/sync/git-exec.ts | 89 +++++++++++++++++ src/sync/manager.test.ts | 114 +++++++++++++++++++++ src/sync/manager.ts | 141 ++++++++++++++++++++++++++ src/sync/planner.test.ts | 99 +++++++++++++++++++ src/sync/planner.ts | 57 +++++++++++ 22 files changed, 1364 insertions(+), 16 deletions(-) create mode 100644 e2e/sync.test.ts create mode 100644 src/commands/vault-sync.test.ts create mode 100644 src/commands/vault-sync.ts create mode 100644 src/sync/conflict.test.ts create mode 100644 src/sync/conflict.ts create mode 100644 src/sync/cycle.test.ts create mode 100644 src/sync/cycle.ts create mode 100644 src/sync/git-exec.test.ts create mode 100644 src/sync/git-exec.ts create mode 100644 src/sync/manager.test.ts create mode 100644 src/sync/manager.ts create mode 100644 src/sync/planner.test.ts create mode 100644 src/sync/planner.ts diff --git a/CLAUDE.md b/CLAUDE.md index de139cd..ed0f8dc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,7 +2,7 @@ The agentage CLI. Versioning RESTARTED at 0.0.x (old npm versions were unpublished; earliest burned slot is 0.1.19 - stay below it until the line naturally -passes). Commands: `setup` (OAuth sign-in), `status`, `vault`, `memory`, `daemon`. The old +passes). Commands: `setup` (OAuth sign-in), `status`, `vault` (incl. `vault sync`), `memory`, `daemon`. The old agent-runtime CLI (run/agents/machines/...) lives in git history only - do not resurrect its agent-runtime patterns (the local memory daemon was deliberately ported from it). @@ -19,6 +19,12 @@ 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/package-guard.test.ts` - CI guard: no agent-runtime remnants (express/ws/sqlite/ core/platform/supabase), runtime deps stay exactly `@agentage/memory-core + chalk + commander + open` diff --git a/e2e/sync.test.ts b/e2e/sync.test.ts new file mode 100644 index 0000000..3a0b4ed --- /dev/null +++ b/e2e/sync.test.ts @@ -0,0 +1,170 @@ +import { execFileSync } from 'node:child_process'; +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { expect, test } from '@playwright/test'; +import { assertCliBuilt, createCliMachine, freePort, type CliMachine } from './helpers.js'; + +// M4 git sync @p0: a vault with an external git origin syncs to a temp bare remote on disk. Fully +// self-contained - a local `git init --bare`, an isolated config dir, and an ephemeral daemon +// port; never the real ~/.agentage, never :4243, no deployed stack. Proves: the daemon auto-loop +// pushes a write within one interval; divergence keeps both sides (.conflict.md, zero loss); an +// unreachable remote never blocks a write. + +const GIT_ENV = { + GIT_AUTHOR_NAME: 'e2e', + GIT_AUTHOR_EMAIL: 'e2e@example.com', + GIT_COMMITTER_NAME: 'e2e', + GIT_COMMITTER_EMAIL: 'e2e@example.com', + GIT_TERMINAL_PROMPT: '0', +}; + +const git = (cwd: string, args: string[]): string => + execFileSync('git', args, { + cwd, + encoding: 'utf8', + // Capture stderr instead of inheriting it, so a probe miss (e.g. cat-file before the first + // push) does not spam the test log. + stdio: ['ignore', 'pipe', 'pipe'], + env: { ...process.env, ...GIT_ENV }, + }); + +// Write vaults.json directly into the isolated config dir for full control over path + origin + +// interval (the CLI `vault add --git` would default the path into the real home). +const writeVaultsConfig = (m: CliMachine, vault: string, path: string, origin: object): void => { + const config = { + version: 1, + default: vault, + vaults: { [vault]: { path, mcp: ['local'], origin: [origin] } }, + }; + writeFileSync(join(m.configDir, 'vaults.json'), JSON.stringify(config, null, 2), 'utf-8'); +}; + +const bareHas = (bare: string, ref: string): boolean => { + try { + git(bare, ['cat-file', '-e', ref]); + return true; + } catch { + return false; + } +}; + +const alive = (pid: number): boolean => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +}; + +test.describe('git sync @p0', () => { + test.beforeAll(() => assertCliBuilt()); + + test('the daemon auto-loop pushes a write to the bare remote within one interval', async () => { + const port = await freePort(); + const m = createCliMachine({ AGENTAGE_NO_DAEMON: '', AGENTAGE_DAEMON_PORT: String(port) }); + const scratch = mkdtempSync(join(tmpdir(), 'agentage-sync-a-')); + const bare = join(scratch, 'remote.git'); + const vaultDir = join(scratch, 'vault'); + let daemonPid: number | null = null; + try { + git(scratch, ['init', '--bare', '-b', 'main', bare]); + writeVaultsConfig(m, 'main', vaultDir, { remote: bare, interval: 1 }); + + const start = await m.exec(['daemon', 'start']); + expect(start.code, start.stderr).toBe(0); + const pidFile = join(m.configDir, 'daemon.pid'); + daemonPid = existsSync(pidFile) + ? Number.parseInt(readFileSync(pidFile, 'utf-8').trim(), 10) + : null; + + const write = await m.exec(['memory', 'write', 'notes/hi.md', '--body', 'sync me quokka']); + expect(write.code, write.stderr).toBe(0); + + // The 1s auto-loop should push well within this window. + const deadline = Date.now() + 30_000; + while (!bareHas(bare, 'main') && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 500)); + } + expect(bareHas(bare, 'main'), 'bare remote received the commit').toBe(true); + expect(git(bare, ['show', 'main:notes/hi.md'])).toContain('sync me quokka'); + + const status = await m.exec(['daemon', 'status']); + expect(status.stdout).toContain('sync'); + + await m.exec(['daemon', 'stop']); + } finally { + if (daemonPid !== null && alive(daemonPid)) { + try { + process.kill(daemonPid, 'SIGKILL'); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== 'ESRCH') throw err; + } + } + m.cleanup(); + rmSync(scratch, { recursive: true, force: true }); + } + }); + + test('divergence keeps both sides: .conflict.md holds the remote copy, zero lost writes', async () => { + // interval 0 (manual-only) + helpers default the daemon off -> a fully in-process forced sync. + const m = createCliMachine(); + const scratch = mkdtempSync(join(tmpdir(), 'agentage-sync-b-')); + const bare = join(scratch, 'remote.git'); + const vaultDir = join(scratch, 'vault'); + try { + git(scratch, ['init', '--bare', '-b', 'main', bare]); + writeVaultsConfig(m, 'main', vaultDir, { remote: bare, interval: 0 }); + + // Seed a base note and push it to the bare remote. + expect((await m.exec(['memory', 'write', 'note.md', '--body', 'base'])).code).toBe(0); + expect((await m.exec(['vault', 'sync', 'main'])).code).toBe(0); + + // A second clone advances the remote with a conflicting change. + const clone = join(scratch, 'clone'); + git(scratch, ['clone', bare, clone]); + writeFileSync(join(clone, 'note.md'), 'REMOTE-CHANGE\n', 'utf-8'); + git(clone, ['commit', '-am', 'remote change']); + git(clone, ['push', 'origin', 'HEAD:main']); + + // Local makes its own conflicting change, then forces a sync. + expect((await m.exec(['memory', 'write', 'note.md', '--body', 'LOCAL-CHANGE'])).code).toBe(0); + const sync = await m.exec(['vault', 'sync', 'main']); + expect(sync.code, sync.stderr).toBe(0); + expect(sync.stdout).toContain('conflict copy'); + + expect(readFileSync(join(vaultDir, 'note.md'), 'utf-8')).toContain('LOCAL-CHANGE'); + expect(readFileSync(join(vaultDir, 'note.conflict.md'), 'utf-8')).toContain('REMOTE-CHANGE'); + // The push landed - both sides live on the remote too. + expect(git(bare, ['show', 'main:note.md'])).toContain('LOCAL-CHANGE'); + expect(git(bare, ['show', 'main:note.conflict.md'])).toContain('REMOTE-CHANGE'); + } finally { + m.cleanup(); + rmSync(scratch, { recursive: true, force: true }); + } + }); + + test('an unreachable remote never blocks a write; the sync error is recorded, not a crash', async () => { + const m = createCliMachine(); + const scratch = mkdtempSync(join(tmpdir(), 'agentage-sync-c-')); + const vaultDir = join(scratch, 'vault'); + try { + writeVaultsConfig(m, 'main', vaultDir, { remote: join(scratch, 'nope.git'), interval: 0 }); + + // The write succeeds even though the remote is unreachable. + const write = await m.exec(['memory', 'write', 'note.md', '--body', 'offline write']); + expect(write.code, write.stderr).toBe(0); + + const sync = await m.exec(['vault', 'sync', 'main']); + expect(sync.code, sync.stderr).toBe(0); // recorded, not a crash + expect(sync.stdout).toContain('unreachable'); + // The local commit persisted - CRUD never blocks on sync. + const log = git(vaultDir, ['log', '--oneline']); + expect(log.trim().length).toBeGreaterThan(0); + } finally { + m.cleanup(); + rmSync(scratch, { recursive: true, force: true }); + } + }); +}); diff --git a/package-lock.json b/package-lock.json index 9cc25db..54095e2 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,9 +9,9 @@ "version": "0.0.3", "license": "MIT", "dependencies": { - "@agentage/memory-core": "^0.1.1", - "@agentage/server-memory": "0.0.2", - "@modelcontextprotocol/sdk": "1.29.0", + "@agentage/memory-core": "^0.1.2", + "@agentage/server-memory": "^0.0.2", + "@modelcontextprotocol/sdk": "^1.29.0", "chalk": "latest", "commander": "latest", "open": "latest" @@ -39,9 +39,9 @@ } }, "node_modules/@agentage/memory-core": { - "version": "0.1.1", - "resolved": "https://registry.npmjs.org/@agentage/memory-core/-/memory-core-0.1.1.tgz", - "integrity": "sha512-e4NMCWIjKoZcwDaemLqSmd02/VgDBEVrRz2/0t9T94O3ERlZhhUjOn0+Ya193YLyjfquRNdVMkXpKHV2k7SE6w==", + "version": "0.1.2", + "resolved": "https://registry.npmjs.org/@agentage/memory-core/-/memory-core-0.1.2.tgz", + "integrity": "sha512-bnUjYcDpvzzPfJMY1ZZ3yOHAdAIku0Ez9f308QjL6Uy36EjllRUPYyfVZRjnXbn4mPp6GMlCCo4wFDEIcxqkFA==", "license": "MIT", "dependencies": { "yaml": "^2.7.0", diff --git a/package.json b/package.json index 9604b4d..8233127 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,7 @@ "prepublishOnly": "npm run verify" }, "dependencies": { - "@agentage/memory-core": "^0.1.1", + "@agentage/memory-core": "^0.1.2", "@agentage/server-memory": "^0.0.2", "@modelcontextprotocol/sdk": "^1.29.0", "chalk": "latest", diff --git a/src/commands/daemon-cmd.ts b/src/commands/daemon-cmd.ts index 5ffda3a..ac6a987 100644 --- a/src/commands/daemon-cmd.ts +++ b/src/commands/daemon-cmd.ts @@ -1,7 +1,7 @@ import chalk from 'chalk'; import { type Command } from 'commander'; import { isDaemonRunning, resolvePort, stopDaemon } from '../daemon/lifecycle.js'; -import { health, mismatchNotice, spawnDaemon } from '../lib/daemon-client.js'; +import { health, mismatchNotice, spawnDaemon, syncStatus } from '../lib/daemon-client.js'; const startAction = async (): Promise => { const port = resolvePort(); @@ -46,6 +46,22 @@ const statusAction = async (): Promise => { console.log(`version ${h.version}`); const notice = mismatchNotice(h.version); if (notice) console.log(chalk.yellow(notice)); + + const sync = await syncStatus(port); + if (sync && sync.vaults.length > 0) { + console.log('sync'); + for (const v of sync.vaults) { + const cadence = v.intervalSeconds > 0 ? `every ${v.intervalSeconds}s` : 'manual'; + const state = v.running + ? 'running' + : v.lastError + ? `error: ${v.lastError}` + : v.lastRun + ? `ok ${v.lastRun}` + : 'scheduled'; + console.log(` ${v.vault.padEnd(16)} ${cadence.padEnd(12)} ${state}`); + } + } }; export const registerDaemon = (program: Command): void => { diff --git a/src/commands/vault-sync.test.ts b/src/commands/vault-sync.test.ts new file mode 100644 index 0000000..e4d2a0c --- /dev/null +++ b/src/commands/vault-sync.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, it, vi } from 'vitest'; +import { type VaultsConfig } from '@agentage/memory-core'; +import { type SyncResult } from '../sync/cycle.js'; +import { type SyncTarget } from '../sync/planner.js'; +import { runVaultSync, type VaultSyncDeps } from './vault-sync.js'; + +const result = (over: Partial = {}): SyncResult => ({ + vault: 'v', + remote: 'git@h:v.git', + ok: true, + committed: false, + pushed: true, + conflicts: [], + ...over, +}); + +const makeDeps = (over: Partial = {}): { deps: VaultSyncDeps; logs: string[] } => { + const logs: string[] = []; + const deps: VaultSyncDeps = { + loadConfig: (): VaultsConfig => ({ + version: 1, + vaults: { v: { path: '/tmp/v', origin: [{ remote: 'git@h:v.git', interval: 0 }] } }, + }), + daemonPort: async () => null, + runViaDaemon: async () => result(), + runInProcess: async () => result(), + log: (m) => logs.push(m), + ...over, + }; + return { deps, logs }; +}; + +describe('runVaultSync', () => { + it('reports "no git origin" for an unknown vault name', async () => { + const { deps, logs } = makeDeps(); + await runVaultSync('missing', deps); + expect(logs.join()).toContain("No git origin configured for vault 'missing'"); + }); + + it('hints when no git-synced vaults exist', async () => { + const { deps, logs } = makeDeps({ loadConfig: () => ({ version: 1, vaults: {} }) }); + await runVaultSync(undefined, deps); + expect(logs.join()).toContain('No git-synced vaults'); + }); + + it('runs in-process when the daemon is down', async () => { + const runInProcess = vi.fn(async (_t: SyncTarget) => result({ committed: true })); + const { deps, logs } = makeDeps({ daemonPort: async () => null, runInProcess }); + await runVaultSync('v', deps); + expect(runInProcess).toHaveBeenCalledTimes(1); + expect(logs.join()).toContain('committed'); + expect(logs.join()).toContain('pushed'); + }); + + it('delegates to the daemon when one is reachable', async () => { + const runViaDaemon = vi.fn(async () => result()); + const runInProcess = vi.fn(async (t: SyncTarget) => result({ vault: t.vault })); + const { deps } = makeDeps({ daemonPort: async () => 4243, runViaDaemon, runInProcess }); + await runVaultSync('v', deps); + expect(runViaDaemon).toHaveBeenCalledWith(4243, 'v'); + expect(runInProcess).not.toHaveBeenCalled(); + }); + + it('surfaces a failure line and lists preserved conflict copies', async () => { + const { deps, logs } = makeDeps({ + runInProcess: async () => + result({ ok: false, pushed: false, reason: 'unreachable', error: 'nope' }), + }); + await runVaultSync('v', deps); + expect(logs.join()).toContain('failed (unreachable)'); + + const { deps: d2, logs: l2 } = makeDeps({ + runInProcess: async () => result({ conflicts: ['note.conflict.md'] }), + }); + await runVaultSync('v', d2); + expect(l2.join()).toContain('kept remote copy: note.conflict.md'); + }); +}); diff --git a/src/commands/vault-sync.ts b/src/commands/vault-sync.ts new file mode 100644 index 0000000..094f037 --- /dev/null +++ b/src/commands/vault-sync.ts @@ -0,0 +1,72 @@ +import chalk from 'chalk'; +import { type VaultsConfig } from '@agentage/memory-core'; +import { health, syncRun } 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 { syncTargets, type SyncTarget } from '../sync/planner.js'; + +export interface VaultSyncDeps { + loadConfig: () => VaultsConfig; + // The port of a reachable daemon, or null to run in-process (daemon down or --no-daemon). + daemonPort: () => Promise; + runViaDaemon: (port: number, vault: string) => Promise; + runInProcess: (target: SyncTarget) => Promise; + log: (msg: string) => void; +} + +const describe = (r: SyncResult): string => { + if (!r.ok) return chalk.red(`failed (${r.reason ?? 'error'})${r.error ? `: ${r.error}` : ''}`); + if (r.skipped) return chalk.yellow(`skipped (${r.skipped})`); + const bits: string[] = []; + if (r.committed) bits.push('committed'); + if (r.conflicts.length) bits.push(`${r.conflicts.length} conflict copy(ies)`); + if (r.pushed) bits.push('pushed'); + return chalk.green(bits.length ? bits.join(', ') : 'up to date'); +}; + +const report = (log: (msg: string) => void, r: SyncResult): void => { + log(`${r.vault} -> ${r.remote}: ${describe(r)}`); + for (const c of r.conflicts) log(` kept remote copy: ${c}`); +}; + +// `agentage vault sync [name]`: sync one vault (or every origin-carrying vault). Prefers a running +// daemon (single writer), else runs the cycle in-process. Works for interval-0 (manual-only) +// vaults and with the daemon down. Sync failures are surfaced, not thrown (V6: never a crash). +export const runVaultSync = async ( + name: string | undefined, + deps: VaultSyncDeps +): Promise => { + const targets = syncTargets(deps.loadConfig()).filter((t) => !name || t.vault === name); + if (targets.length === 0) { + deps.log( + name + ? `No git origin configured for vault '${name}'.` + : 'No git-synced vaults. Add one with `agentage vault add --git `.' + ); + return; + } + const port = await deps.daemonPort(); + if (port !== null) { + for (const vault of [...new Set(targets.map((t) => t.vault))]) { + report(deps.log, await deps.runViaDaemon(port, vault)); + } + return; + } + for (const target of targets) report(deps.log, await deps.runInProcess(target)); +}; + +const resolveDaemonPort = async (): Promise => { + if (daemonDisabled()) return null; + const port = resolvePort(); + return (await health(port)) ? port : null; +}; + +export const defaultVaultSyncDeps = (): VaultSyncDeps => ({ + loadConfig: () => loadVaultsConfig().config, + daemonPort: resolveDaemonPort, + runViaDaemon: syncRun, + runInProcess: runSyncCycle, + log: (msg) => console.log(msg), +}); diff --git a/src/commands/vault.test.ts b/src/commands/vault.test.ts index f5f3cf6..4262746 100644 --- a/src/commands/vault.test.ts +++ b/src/commands/vault.test.ts @@ -35,11 +35,17 @@ describe('vault add', () => { expect(h.ensured).toEqual(['~/vaults/notes']); }); - it('registers a --git vault as an origin (no dir created)', () => { + it('registers a --git vault as a local working copy synced to an origin', () => { const h = makeDeps(); runVaultAdd('work', { git: 'git@github.com:me/w.git' }, h.deps); - expect(h.get().vaults?.work.origin?.[0]?.remote).toBe('git@github.com:me/w.git'); - expect(h.ensured).toEqual([]); + expect(h.get().vaults?.work).toEqual({ + path: '~/vaults/work', + origin: [{ remote: 'git@github.com:me/w.git' }], + mcp: ['local'], + }); + // The working copy dir is created so the daemon has somewhere to commit/push from. + expect(h.ensured).toEqual(['~/vaults/work']); + expect(h.logs.join()).toContain('(git)'); }); it('rejects add with no flag', () => { diff --git a/src/commands/vault.ts b/src/commands/vault.ts index c74c704..38771df 100644 --- a/src/commands/vault.ts +++ b/src/commands/vault.ts @@ -3,6 +3,7 @@ import { type Command } from 'commander'; import { type VaultEntry, type VaultsConfig } from '@agentage/memory-core'; import { addVault, ensureVaultDir, formatVaultLine, removeVault } from '../lib/vault-registry.js'; import { loadVaultsConfig, saveVaultsConfig, type LoadedVaults } from '../lib/vaults.js'; +import { defaultVaultSyncDeps, runVaultSync } from './vault-sync.js'; export interface VaultDeps { load: () => LoadedVaults; @@ -27,7 +28,9 @@ export interface VaultAddOptions { const buildEntry = (name: string, opts: VaultAddOptions): VaultEntry => { const hasLocal = opts.local !== undefined; if (hasLocal && opts.git) throw new Error('choose one of --local or --git, not both'); - if (opts.git) return { origin: [{ remote: opts.git }], mcp: ['local'] }; + // A --git vault is a local working copy that syncs to an external git remote (path + origin); + // the daemon commits/pushes and pulls it per its interval. + if (opts.git) return { path: `~/vaults/${name}`, origin: [{ remote: opts.git }], mcp: ['local'] }; if (hasLocal) { const path = typeof opts.local === 'string' ? opts.local : `~/vaults/${name}`; return { path, mcp: ['local'] }; @@ -44,9 +47,10 @@ export const runVaultAdd = ( const config = addVault(deps.load().config, name, entry); if (entry.path) deps.ensureDir(entry.path); deps.save(config); - const kind = entry.path ? 'local' : 'remote'; + const kind = entry.path ? (entry.origin?.length ? 'git' : 'local') : 'remote'; const where = entry.path ?? entry.origin?.[0]?.remote ?? ''; - deps.log(chalk.green(`Added vault '${name}' (${kind}) -> ${where}`)); + const via = entry.path && entry.origin?.length ? ` <- ${entry.origin[0]!.remote}` : ''; + deps.log(chalk.green(`Added vault '${name}' (${kind}) -> ${where}${via}`)); }; export const runVaultRemove = (name: string, deps: VaultDeps = defaultDeps): void => { @@ -105,4 +109,14 @@ export const registerVault = (program: Command): void => { .command('remove ') .description('Unregister a vault (files stay on disk)') .action((name: string) => guard(() => runVaultRemove(name))); + + vault + .command('sync [name]') + .description('Sync git-synced vaults now (commit + push, pull-rebase)') + .action((name: string | undefined) => + runVaultSync(name, defaultVaultSyncDeps()).catch((err: unknown) => { + console.error(chalk.red(err instanceof Error ? err.message : String(err))); + process.exitCode = 1; + }) + ); }; diff --git a/src/daemon-entry.ts b/src/daemon-entry.ts index 2f305b7..25cdd63 100644 --- a/src/daemon-entry.ts +++ b/src/daemon-entry.ts @@ -1,3 +1,4 @@ +import { unwatchFile, watchFile } from 'node:fs'; import { createClientProvider } from './daemon/client-provider.js'; import { removePidFile, @@ -8,22 +9,33 @@ import { } from './daemon/lifecycle.js'; import { createDaemonServer } from './daemon/server.js'; import { loadLocalMemoryServer } from './mcp/local-server.js'; +import { vaultsJsonPath } from './lib/vaults.js'; +import { createSyncManager } from './sync/manager.js'; import { VERSION } from './utils/version.js'; // The detached, long-lived engine host: one loopback HTTP server that owns a single in-process -// engine and serialises every vault mutation, avoiding concurrent git index.lock collisions. +// engine and serialises every vault mutation, avoiding concurrent git index.lock collisions. It +// also runs the git-sync loop (per-vault origin push/pull) and reschedules on config change. const main = async (): Promise => { const port = resolvePort(); + const manager = createSyncManager(); const server = createDaemonServer({ getClient: createClientProvider(), buildMcpServer: loadLocalMemoryServer, + sync: { status: () => manager.status(), runNow: (vault) => manager.runNow(vault) }, version: VERSION, }); await server.start(port); writePidFile(process.pid); writePortFile(port); + manager.reschedule(); + + const configPath = vaultsJsonPath(); + watchFile(configPath, { interval: 2000 }, () => manager.reschedule()); const shutdown = (): void => { + unwatchFile(configPath); + manager.stop(); server.stop().finally(() => { removePidFile(); removePortFile(); diff --git a/src/daemon/server.ts b/src/daemon/server.ts index 2b9b674..df16d7d 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -1,15 +1,24 @@ 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 SyncStatus } from '../sync/manager.js'; import { dispatchMemory, isMemoryVerb } from './actions.js'; import { handleMcp } from './mcp-http.js'; const LOOPBACK = '127.0.0.1'; +export interface DaemonSyncApi { + status: () => SyncStatus; + runNow: (vault: string) => Promise; +} + export interface DaemonServerOptions { getClient: () => MemoryClient | Promise; // Builds a fresh MCP server per request for POST /mcp; omit to leave the endpoint unmounted. buildMcpServer?: () => Promise; + // The git-sync surface: GET /api/sync/status + POST /api/sync/run; omit to leave both unmounted. + sync?: DaemonSyncApi; version: string; startedAt?: number; } @@ -63,6 +72,20 @@ export const createDaemonServer = (opts: DaemonServerOptions): DaemonServer => { if (!opts.buildMcpServer) return send(res, 404, { error: 'not found' }); return handleMcp(req, res, opts.buildMcpServer); } + if (req.method === 'GET' && url === '/api/sync/status') { + if (!opts.sync) return send(res, 404, { error: 'not found' }); + return send(res, 200, opts.sync.status()); + } + if (req.method === 'POST' && url === '/api/sync/run') { + if (!opts.sync) return send(res, 404, { error: 'not found' }); + const body = (await readBody(req)) as { vault?: string }; + if (!body.vault) return send(res, 400, { error: 'vault is required' }); + try { + return send(res, 200, await opts.sync.runNow(body.vault)); + } catch (err) { + return send(res, 400, { error: err instanceof Error ? err.message : String(err) }); + } + } const match = url.match(/^\/api\/memory\/([a-z]+)$/); if (req.method === 'POST' && match) { const verb = match[1]; diff --git a/src/lib/daemon-client.ts b/src/lib/daemon-client.ts index 5f2b480..40121ba 100644 --- a/src/lib/daemon-client.ts +++ b/src/lib/daemon-client.ts @@ -9,6 +9,8 @@ import { type WriteResult, } from '@agentage/memory-core'; import { resolvePort } from '../daemon/lifecycle.js'; +import { type SyncResult } from '../sync/cycle.js'; +import { type SyncStatus } from '../sync/manager.js'; import { VERSION } from '../utils/version.js'; import { type DeleteResult, @@ -52,6 +54,33 @@ export const health = async (port: number, timeoutMs = 1000): Promise => { + try { + const res = await fetch(`${base(port)}/api/sync/status`, { + signal: AbortSignal.timeout(timeoutMs), + }); + return res.ok ? ((await res.json()) as SyncStatus) : null; + } catch { + return null; + } +}; + +// Ask the daemon to sync one vault now; the daemon runs the cycle in its own process. +export const syncRun = async (port: number, vault: string): Promise => { + const res = await fetch(`${base(port)}/api/sync/run`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ vault }), + }); + if (!res.ok) { + const data = (await res.json().catch(() => ({}))) as { error?: string }; + throw new Error(data.error || `sync request failed: ${res.status}`); + } + return res.json() as Promise; +}; + export const waitForHealth = async ( port: number, opts: { timeoutMs?: number; intervalMs?: number } = {} diff --git a/src/sync/conflict.test.ts b/src/sync/conflict.test.ts new file mode 100644 index 0000000..785c9d4 --- /dev/null +++ b/src/sync/conflict.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from 'vitest'; +import { conflictName } from './conflict.js'; + +const none = (): boolean => false; + +describe('conflictName', () => { + it('inserts .conflict before the extension', () => { + expect(conflictName('notes/q.md', none)).toBe('notes/q.conflict.md'); + }); + + it('handles a path with no extension', () => { + expect(conflictName('README', none)).toBe('README.conflict'); + }); + + it('does not treat a dot in a folder name as the extension', () => { + expect(conflictName('a.b/c', none)).toBe('a.b/c.conflict'); + }); + + it('does not clobber an existing conflict file - suffixes instead', () => { + const taken = new Set(['notes/q.conflict.md', 'notes/q.conflict-1.md']); + expect(conflictName('notes/q.md', (c) => taken.has(c))).toBe('notes/q.conflict-2.md'); + }); + + it('leaves a leading-dot filename intact', () => { + expect(conflictName('.obsidianrc', none)).toBe('.obsidianrc.conflict'); + }); +}); diff --git a/src/sync/conflict.ts b/src/sync/conflict.ts new file mode 100644 index 0000000..b992066 --- /dev/null +++ b/src/sync/conflict.ts @@ -0,0 +1,14 @@ +// The name for the file that preserves the remote side of a conflict: `x.md` -> `x.conflict.md`. +// An existing name is never clobbered - a numeric suffix is added instead (`x.conflict-1.md`). +export const conflictName = (path: string, exists: (candidate: string) => boolean): string => { + const slash = path.lastIndexOf('/'); + const dot = path.lastIndexOf('.'); + const hasExt = dot > slash + 1; // a dot inside the basename, not a leading dot + const stem = hasExt ? path.slice(0, dot) : path; + const ext = hasExt ? path.slice(dot) : ''; + const base = `${stem}.conflict${ext}`; + if (!exists(base)) return base; + let i = 1; + while (exists(`${stem}.conflict-${i}${ext}`)) i += 1; + return `${stem}.conflict-${i}${ext}`; +}; diff --git a/src/sync/cycle.test.ts b/src/sync/cycle.test.ts new file mode 100644 index 0000000..f918f60 --- /dev/null +++ b/src/sync/cycle.test.ts @@ -0,0 +1,158 @@ +import { execFileSync } from 'node:child_process'; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { dirname, join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { runSyncCycle } from './cycle.js'; +import { type SyncTarget } from './planner.js'; + +// Real git in temp dirs: a bare remote + a working copy, exercising commit/push, no-op, ignore, +// conflict materialization, and unreachable handling end to end (the unit-tier proof of the sync +// engine's git behavior; the pure planner/conflict logic is covered separately). + +const IDENTITY = { + GIT_AUTHOR_NAME: 'test', + GIT_AUTHOR_EMAIL: 'test@example.com', + GIT_COMMITTER_NAME: 'test', + GIT_COMMITTER_EMAIL: 'test@example.com', + GIT_TERMINAL_PROMPT: '0', +}; + +const g = (cwd: string, args: string[]): string => + execFileSync('git', args, { cwd, encoding: 'utf8', env: { ...process.env, ...IDENTITY } }); + +const writeFile = (root: string, rel: string, body: string): void => { + const abs = join(root, rel); + mkdirSync(dirname(abs), { recursive: true }); + writeFileSync(abs, body, 'utf8'); +}; + +const target = (over: Partial & Pick): SyncTarget => ({ + vault: 'v', + remoteName: 'sync', + intervalSeconds: 0, + ignore: ['.obsidian/', 'data.json'], + ...over, +}); + +describe('runSyncCycle', () => { + let root: string; + let bare: string; + let work: string; + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'sync-cycle-')); + bare = join(root, 'remote.git'); + work = join(root, 'work'); + g(root, ['init', '--bare', '-b', 'main', bare]); + mkdirSync(work); + }); + afterEach(() => rmSync(root, { recursive: true, force: true })); + + it('commits a dirty working tree and pushes it to an empty bare remote', async () => { + writeFile(work, 'notes/a.md', 'hello quokka'); + const result = await runSyncCycle(target({ path: work, remote: bare })); + expect(result.ok).toBe(true); + expect(result.committed).toBe(true); + expect(result.pushed).toBe(true); + expect(g(bare, ['show', 'main:notes/a.md'])).toContain('hello quokka'); + }); + + it('is a no-op on the second cycle (nothing to commit)', async () => { + writeFile(work, 'a.md', 'x'); + await runSyncCycle(target({ path: work, remote: bare })); + const second = await runSyncCycle(target({ path: work, remote: bare })); + expect(second.ok).toBe(true); + expect(second.committed).toBe(false); + expect(second.pushed).toBe(true); + }); + + it('honors the ignore list (defaults exclude .obsidian/ and data.json)', async () => { + writeFile(work, 'note.md', 'keep me'); + writeFile(work, '.obsidian/app.json', '{}'); + writeFile(work, 'data.json', '{}'); + const result = await runSyncCycle(target({ path: work, remote: bare })); + expect(result.ok).toBe(true); + const tracked = g(work, ['ls-files']).trim().split('\n'); + expect(tracked).toContain('note.md'); + expect(tracked).not.toContain('.obsidian/app.json'); + expect(tracked).not.toContain('data.json'); + }); + + it('an empty ignore list syncs everything', async () => { + writeFile(work, 'note.md', 'keep me'); + writeFile(work, '.obsidian/app.json', '{}'); + const result = await runSyncCycle(target({ path: work, remote: bare, ignore: [] })); + expect(result.ok).toBe(true); + const tracked = g(work, ['ls-files']).trim().split('\n'); + expect(tracked).toContain('note.md'); + expect(tracked).toContain('.obsidian/app.json'); + }); + + it('keeps both sides on divergence: local file stays, remote copy -> .conflict.md, zero loss', async () => { + // Shared base pushed to the bare remote. + writeFile(work, 'note.md', 'base\n'); + await runSyncCycle(target({ path: work, remote: bare })); + + // A second clone advances the remote with a conflicting change. + const other = join(root, 'other'); + g(root, ['clone', bare, other]); + writeFile(other, 'note.md', 'REMOTE-CHANGE\n'); + g(other, ['commit', '-am', 'remote change']); + g(other, ['push', 'origin', 'HEAD:main']); + + // Local makes its own conflicting change (as the engine would: a committed mutation). + writeFile(work, 'note.md', 'LOCAL-CHANGE\n'); + g(work, ['commit', '-am', 'local change']); + + const result = await runSyncCycle(target({ path: work, remote: bare })); + expect(result.ok).toBe(true); + expect(result.pushed).toBe(true); + expect(result.conflicts).toEqual(['note.conflict.md']); + + // Local: original file keeps the local side; the conflict file holds the remote side. + expect(readFileSync(join(work, 'note.md'), 'utf8')).toContain('LOCAL-CHANGE'); + expect(readFileSync(join(work, 'note.conflict.md'), 'utf8')).toContain('REMOTE-CHANGE'); + + // Remote: the push landed, both sides preserved there too. + expect(g(bare, ['show', 'main:note.md'])).toContain('LOCAL-CHANGE'); + expect(g(bare, ['show', 'main:note.conflict.md'])).toContain('REMOTE-CHANGE'); + }); + + it('merges remote non-conflicting changes cleanly', async () => { + writeFile(work, 'a.md', 'a\n'); + await runSyncCycle(target({ path: work, remote: bare })); + + const other = join(root, 'other'); + g(root, ['clone', bare, other]); + writeFile(other, 'b.md', 'from remote\n'); + g(other, ['add', '-A']); + g(other, ['commit', '-m', 'add b']); + g(other, ['push', 'origin', 'HEAD:main']); + + writeFile(work, 'a.md', 'a changed\n'); + g(work, ['commit', '-am', 'change a']); + + const result = await runSyncCycle(target({ path: work, remote: bare })); + expect(result.ok).toBe(true); + expect(result.conflicts).toEqual([]); + // Remote's non-conflicting file is now present locally (no data lost). + expect(existsSync(join(work, 'b.md'))).toBe(true); + }); + + it('records an unreachable remote without crashing', async () => { + writeFile(work, 'a.md', 'x'); + const result = await runSyncCycle( + target({ path: work, remote: join(root, 'does-not-exist.git') }) + ); + expect(result.ok).toBe(false); + expect(result.reason).toBe('unreachable'); + // The local commit still happened - CRUD never blocks on sync. + expect(result.committed).toBe(true); + expect(g(work, ['log', '--oneline']).trim().length).toBeGreaterThan(0); + }); + + it('returns a clean no-op when the vault directory does not exist yet', async () => { + const result = await runSyncCycle(target({ path: join(root, 'ghost'), remote: bare })); + expect(result).toMatchObject({ ok: true, committed: false, pushed: false }); + }); +}); diff --git a/src/sync/cycle.ts b/src/sync/cycle.ts new file mode 100644 index 0000000..788f624 --- /dev/null +++ b/src/sync/cycle.ts @@ -0,0 +1,168 @@ +import { existsSync } from 'node:fs'; +import { mkdir, writeFile } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import { conflictName } from './conflict.js'; +import { createSyncGit, GitError, type GitErrorKind, type SyncGit } from './git-exec.js'; +import { type SyncTarget } from './planner.js'; + +export interface SyncResult { + vault: string; + remote: string; + ok: boolean; + committed: boolean; // a sync commit was made for a dirty working tree + pushed: boolean; + conflicts: string[]; // paths written as `.conflict.md` + skipped?: 'lock' | 'busy'; + reason?: GitErrorKind; + error?: string; +} + +export interface CycleDeps { + git?: SyncGit; + now?: () => string; +} + +const branchOf = async (git: SyncGit): Promise => { + const out = await git.exec(['symbolic-ref', '--short', 'HEAD']); + return out.code === 0 && out.stdout.trim() ? out.stdout.trim() : 'main'; +}; + +// Local, uncommitted, non-ignored changes -> one sync commit. Returns whether a commit was made. +const commitIfDirty = async (git: SyncGit, message: string): Promise => { + await git.run(['add', '-A']); + const staged = await git.exec(['diff', '--cached', '--quiet']); + if (staged.code === 0) return false; + await git.run(['commit', '-m', message]); + return true; +}; + +// The ignore rules ride on `.git/info/exclude` (local, uncommitted, gitignore syntax) rather than +// per-command pathspecs: it is the robust, idempotent mechanism - matching untracked files are +// never staged by `add -A`, and it leaves no committed `.gitignore` in the user's vault. An empty +// list clears it (sync everything). +const writeExclude = async (path: string, patterns: string[]): Promise => { + const infoDir = join(path, '.git', 'info'); + await mkdir(infoDir, { recursive: true }); + await writeFile( + join(infoDir, 'exclude'), + patterns.length ? patterns.join('\n') + '\n' : '', + 'utf8' + ); +}; + +const ensureRemote = async (git: SyncGit, name: string, url: string): Promise => { + const remotes = (await git.exec(['remote'])).stdout.split('\n').map((r) => r.trim()); + if (remotes.includes(name)) await git.run(['remote', 'set-url', name, url]); + else await git.run(['remote', 'add', name, url]); +}; + +// Reconcile a diverged history keeping BOTH sides. A rebase probes for a true conflict; on a clean +// replay history stays linear. On conflict the local files are kept as-is and each conflicted +// file's remote copy is written alongside as `.conflict.md`, so no write is lost on either +// side. Returns the conflict-copy paths written. +const reconcile = async ( + git: SyncGit, + cwd: string, + ref: string, + now: string +): Promise => { + const rebase = await git.exec(['rebase', ref]); + if (rebase.code === 0) return []; + + const conflicted = (await git.exec(['diff', '--name-only', '--diff-filter=U'])).stdout + .split('\n') + .map((p) => p.trim()) + .filter(Boolean); + await git.exec(['rebase', '--abort']); + + const remoteSides = new Map(); + for (const p of conflicted) { + const show = await git.exec(['show', `${ref}:${p}`]); + if (show.code === 0) remoteSides.set(p, show.stdout); + } + + // Merge remote in, auto-resolving content conflicts to the local side; keeps remote's + // non-conflicting changes and yields a push-able (descendant-of-remote) history. + const merge = await git.exec(['merge', '-X', 'ours', '--no-edit', ref]); + if (merge.code !== 0) { + await git.exec(['checkout', '--ours', '--', '.']); + await git.run(['add', '-A']); + await git.run(['commit', '--no-edit']); + } + + const written: string[] = []; + for (const [p, content] of remoteSides) { + const name = conflictName(p, (candidate) => existsSync(join(cwd, candidate))); + const abs = join(cwd, name); + await mkdir(dirname(abs), { recursive: true }); + await writeFile(abs, content, 'utf8'); + written.push(name); + } + if (written.length) { + await git.run(['add', '-A']); + await git.run([ + 'commit', + '-m', + `sync: preserve remote copy of ${written.length} conflicted file(s) (${now})`, + ]); + } + return written; +}; + +// One tolerant sync cycle for a single target: commit-if-dirty -> ensure remote -> fetch -> +// pull-rebase (keeping both sides on conflict) -> push. Every git failure is caught and classified +// so an unreachable remote or a held index.lock is a clean skip that the next cycle catches up. +export const runSyncCycle = async ( + target: SyncTarget, + deps: CycleDeps = {} +): Promise => { + const git = deps.git ?? createSyncGit(target.path); + const now = (deps.now ?? (() => new Date().toISOString()))(); + const base: SyncResult = { + vault: target.vault, + remote: target.remote, + ok: false, + committed: false, + pushed: false, + conflicts: [], + }; + let committed = false; + let conflicts: string[] = []; + try { + if (!existsSync(target.path)) return { ...base, ok: true }; // nothing on disk yet + if (!existsSync(join(target.path, '.git'))) await git.run(['init', '-b', 'main']); + + await writeExclude(target.path, target.ignore); + await ensureRemote(git, target.remoteName, target.remote); + committed = await commitIfDirty(git, `sync: ${now}`); + const branch = await branchOf(git); + const ref = `${target.remoteName}/${branch}`; + + const fetch = await git.exec(['fetch', target.remoteName]); + if (fetch.code !== 0) throw new GitError(fetch); + + if ((await git.exec(['rev-parse', '--verify', ref])).code === 0) { + conflicts = await reconcile(git, target.path, ref, now); + } + + const push = await git.exec(['push', target.remoteName, `HEAD:${branch}`]); + if (push.code !== 0) throw new GitError(push); + + return { ...base, ok: true, committed, pushed: true, conflicts }; + } catch (err) { + if (err instanceof GitError) { + // A concurrent engine mutation holding index.lock -> skip cleanly, retry next cycle. + if (err.kind === 'lock') + return { ...base, ok: true, committed, conflicts, skipped: 'lock', reason: 'lock' }; + return { ...base, ok: false, committed, conflicts, reason: err.kind, error: err.message }; + } + return { + ...base, + ok: false, + committed, + conflicts, + reason: 'other', + error: err instanceof Error ? err.message : String(err), + }; + } +}; diff --git a/src/sync/git-exec.test.ts b/src/sync/git-exec.test.ts new file mode 100644 index 0000000..772d550 --- /dev/null +++ b/src/sync/git-exec.test.ts @@ -0,0 +1,55 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { classifyGitError, createSyncGit, GitError } from './git-exec.js'; + +describe('classifyGitError', () => { + it('detects a held index.lock as a lock error', () => { + expect(classifyGitError("fatal: Unable to create '/v/.git/index.lock': File exists.")).toBe( + 'lock' + ); + expect(classifyGitError('Another git process seems to be running')).toBe('lock'); + }); + + it('detects an unreachable remote', () => { + expect(classifyGitError("fatal: '/nope/x.git' does not appear to be a git repository")).toBe( + 'unreachable' + ); + expect(classifyGitError('ssh: Could not resolve hostname github.com')).toBe('unreachable'); + expect(classifyGitError('Connection refused')).toBe('unreachable'); + }); + + it('detects a merge/rebase conflict', () => { + expect(classifyGitError('CONFLICT (content): Merge conflict in a.md')).toBe('conflict'); + expect(classifyGitError('error: could not apply 1a2b3c...')).toBe('conflict'); + }); + + it('falls back to other for anything else', () => { + expect(classifyGitError('some unrelated failure')).toBe('other'); + }); +}); + +describe('createSyncGit', () => { + let dir: string; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'sync-git-exec-')); + }); + afterEach(() => rmSync(dir, { recursive: true, force: true })); + + it('run returns stdout on success and throws a classified GitError on failure', async () => { + const git = createSyncGit(dir); + await git.run(['init', '-b', 'main']); + const branch = (await git.run(['symbolic-ref', '--short', 'HEAD'])).trim(); + expect(branch).toBe('main'); + + await expect(git.run(['rev-parse', '--verify', 'sync/main'])).rejects.toBeInstanceOf(GitError); + }); + + it('exec never throws and reports a non-zero code', async () => { + const git = createSyncGit(dir); + const res = await git.exec(['fetch', 'sync']); + expect(res.code).not.toBe(0); + expect(res.stderr.length).toBeGreaterThan(0); + }); +}); diff --git a/src/sync/git-exec.ts b/src/sync/git-exec.ts new file mode 100644 index 0000000..8e3b3cf --- /dev/null +++ b/src/sync/git-exec.ts @@ -0,0 +1,89 @@ +import { execFile, type ExecFileException } from 'node:child_process'; + +// A tiny git-exec wrapper scoped to one vault working copy. memory-core keeps its own createGit +// private, so sync spawns git directly (cwd = vault path) with a full, ambient-config-free +// identity (CI runners have none) and no interactive prompts (so an unreachable remote fails fast +// instead of hanging on credentials). + +export type GitErrorKind = 'unreachable' | 'lock' | 'conflict' | 'other'; + +export interface GitRunResult { + code: number; + stdout: string; + stderr: string; +} + +// Classify a git failure by its stderr so callers can react: an unreachable remote and a held +// index.lock are both skip-and-retry (not crashes); a conflict drives the keep-both-sides path. +export const classifyGitError = (stderr: string): GitErrorKind => { + const s = stderr.toLowerCase(); + if (/index\.lock|another git process|unable to create .*\.lock|cannot lock ref/.test(s)) + return 'lock'; + if (/conflict|could not apply|needs merge|automatic merge failed/.test(s)) return 'conflict'; + if ( + /could not read from remote|connection refused|could not resolve host|repository .* not found|does not appear to be a git repository|unable to access|no route to host|network is unreachable|permission denied \(publickey\)|host key verification failed|does not exist|failed to connect|remote end hung up/.test( + s + ) + ) + return 'unreachable'; + return 'other'; +}; + +export class GitError extends Error { + readonly kind: GitErrorKind; + readonly stderr: string; + readonly code: number; + constructor(result: GitRunResult) { + super(result.stderr.trim() || `git exited with code ${result.code}`); + this.name = 'GitError'; + this.kind = classifyGitError(result.stderr); + this.stderr = result.stderr; + this.code = result.code; + } +} + +export interface SyncGit { + // Never throws on a non-zero exit; returns the full result for the caller to branch on. + exec(args: string[], opts?: { timeoutMs?: number }): Promise; + // Throws GitError on a non-zero exit; returns stdout otherwise. + run(args: string[], opts?: { timeoutMs?: number }): Promise; +} + +const IDENTITY: Record = { + GIT_AUTHOR_NAME: 'agentage sync', + GIT_AUTHOR_EMAIL: 'sync@agentage.io', + GIT_COMMITTER_NAME: 'agentage sync', + GIT_COMMITTER_EMAIL: 'sync@agentage.io', + GIT_TERMINAL_PROMPT: '0', + GIT_SSH_COMMAND: 'ssh -oBatchMode=yes', +}; + +export const createSyncGit = (cwd: string): SyncGit => { + const exec = (args: string[], opts: { timeoutMs?: number } = {}): Promise => + new Promise((resolve) => { + execFile( + 'git', + args, + { + cwd, + env: { ...process.env, ...IDENTITY }, + encoding: 'utf8', + maxBuffer: 64 * 1024 * 1024, + timeout: opts.timeoutMs ?? 0, + }, + (err: ExecFileException | null, stdout, stderr) => { + const code = typeof err?.code === 'number' ? err.code : err ? 1 : 0; + resolve({ code, stdout: stdout ?? '', stderr: stderr ?? '' }); + } + ); + }); + + return { + exec, + async run(args, opts) { + const result = await exec(args, opts); + if (result.code !== 0) throw new GitError(result); + return result.stdout; + }, + }; +}; diff --git a/src/sync/manager.test.ts b/src/sync/manager.test.ts new file mode 100644 index 0000000..7724978 --- /dev/null +++ b/src/sync/manager.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { type VaultsConfig } from '@agentage/memory-core'; +import { type SyncResult } from './cycle.js'; +import { createSyncManager } from './manager.js'; +import { type SyncTarget } from './planner.js'; + +const ok = (t: SyncTarget): SyncResult => ({ + vault: t.vault, + remote: t.remote, + ok: true, + committed: false, + pushed: true, + conflicts: [], +}); + +const config = (vaults: NonNullable): VaultsConfig => ({ + version: 1, + vaults, +}); + +describe('createSyncManager', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it('schedules only interval>0 targets and fires them on the interval', async () => { + const runs: string[] = []; + const runCycle = vi.fn(async (t: SyncTarget) => { + runs.push(t.vault); + return ok(t); + }); + const manager = createSyncManager({ + getConfig: () => + config({ + auto: { path: '/tmp/a', origin: [{ remote: 'git@h:a.git', interval: 1 }] }, + manual: { path: '/tmp/m', origin: [{ remote: 'git@h:m.git', interval: 0 }] }, + }), + runCycle, + }); + manager.reschedule(); + + // Both appear in status (manual as scheduled/manual); only auto has a timer. + expect( + manager + .status() + .vaults.map((v) => v.vault) + .sort() + ).toEqual(['auto', 'manual']); + + await vi.advanceTimersByTimeAsync(1_000); + expect(runs).toEqual(['auto']); + manager.stop(); + }); + + it('records last-run and last-error per vault', async () => { + const runCycle = vi.fn(async (t: SyncTarget): Promise => ({ + ...ok(t), + ok: false, + pushed: false, + reason: 'unreachable', + error: 'boom', + })); + const manager = createSyncManager({ + getConfig: () => + config({ v: { path: '/tmp/v', origin: [{ remote: 'git@h:v.git', interval: 0 }] } }), + runCycle, + }); + await manager.runNow('v'); + const state = manager.status().vaults[0]; + expect(state?.lastError).toBe('boom'); + expect(state?.lastResult?.reason).toBe('unreachable'); + expect(state?.lastRun).toBeDefined(); + }); + + it('serialises a target against itself (no overlapping cycles)', async () => { + let release: () => void = () => {}; + const gate = new Promise((r) => (release = r)); + const runCycle = vi.fn(async (t: SyncTarget) => { + await gate; + return ok(t); + }); + const manager = createSyncManager({ + getConfig: () => + config({ v: { path: '/tmp/v', origin: [{ remote: 'git@h:v.git', interval: 0 }] } }), + runCycle, + }); + const first = manager.runNow('v'); + const second = await manager.runNow('v'); // in-flight -> busy skip, does not re-enter runCycle + expect(second.skipped).toBe('busy'); + release(); + await first; + expect(runCycle).toHaveBeenCalledTimes(1); + }); + + it('runNow throws for a vault with no configured origin', async () => { + const manager = createSyncManager({ getConfig: () => config({}) }); + await expect(manager.runNow('nope')).rejects.toThrow(/no sync origin/); + }); + + it('reschedule drops timers/state for vaults removed from the config', async () => { + let vaults: NonNullable = { + v: { path: '/tmp/v', origin: [{ remote: 'git@h:v.git', interval: 5 }] }, + }; + const manager = createSyncManager({ + getConfig: () => config(vaults), + runCycle: async (t) => ok(t), + }); + manager.reschedule(); + expect(manager.status().vaults).toHaveLength(1); + vaults = {}; + manager.reschedule(); + expect(manager.status().vaults).toHaveLength(0); + manager.stop(); + }); +}); diff --git a/src/sync/manager.ts b/src/sync/manager.ts new file mode 100644 index 0000000..0f17497 --- /dev/null +++ b/src/sync/manager.ts @@ -0,0 +1,141 @@ +import { type VaultsConfig } from '@agentage/memory-core'; +import { loadVaultsConfig } from '../lib/vaults.js'; +import { runSyncCycle, type SyncResult } from './cycle.js'; +import { autoSyncTargets, intervalMs, syncTargets, type SyncTarget } from './planner.js'; + +export interface VaultSyncState { + vault: string; + remote: string; + intervalSeconds: number; + running: boolean; + lastRun?: string; + lastError?: string; + lastResult?: Pick; +} + +export interface SyncStatus { + vaults: VaultSyncState[]; +} + +export interface SyncManagerDeps { + getConfig?: () => VaultsConfig; + runCycle?: (target: SyncTarget) => Promise; +} + +export interface SyncManager { + // (Re)build the per-target timers from the current config; call on boot and on config change. + reschedule(): void; + // Force a sync of every origin of one vault now (works for interval 0 and daemon-scheduled alike). + runNow(vault: string): Promise; + status(): SyncStatus; + stop(): void; +} + +const busy = (t: SyncTarget): SyncResult => ({ + vault: t.vault, + remote: t.remote, + ok: true, + committed: false, + pushed: false, + conflicts: [], + skipped: 'busy', +}); + +// The daemon-side scheduler: owns per-target timers, serialises each target against itself (an +// in-process running flag - no overlapping cycles), and records last-run/last-error for status. +export const createSyncManager = (deps: SyncManagerDeps = {}): SyncManager => { + const getConfig = deps.getConfig ?? (() => loadVaultsConfig().config); + const runCycle = deps.runCycle ?? runSyncCycle; + const states = new Map(); + const timers = new Map(); + const keyOf = (t: SyncTarget): string => `${t.vault}::${t.remoteName}`; + + const ensureState = (t: SyncTarget): VaultSyncState => { + const existing = states.get(keyOf(t)); + if (existing) { + existing.remote = t.remote; + existing.intervalSeconds = t.intervalSeconds; + return existing; + } + const fresh: VaultSyncState = { + vault: t.vault, + remote: t.remote, + intervalSeconds: t.intervalSeconds, + running: false, + }; + states.set(keyOf(t), fresh); + return fresh; + }; + + const runTarget = async (t: SyncTarget): Promise => { + const state = ensureState(t); + if (state.running) return busy(t); // serialise: no overlapping cycles per target + state.running = true; + try { + const result = await runCycle(t); + state.lastRun = new Date().toISOString(); + state.lastError = result.ok ? undefined : result.error; + state.lastResult = { + ok: result.ok, + pushed: result.pushed, + committed: result.committed, + conflicts: result.conflicts, + skipped: result.skipped, + reason: result.reason, + }; + return result; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + state.lastRun = new Date().toISOString(); + state.lastError = message; + state.lastResult = { + ok: false, + pushed: false, + committed: false, + conflicts: [], + reason: 'other', + }; + return { + vault: t.vault, + remote: t.remote, + ok: false, + committed: false, + pushed: false, + conflicts: [], + reason: 'other', + error: message, + }; + } finally { + state.running = false; + } + }; + + const runNow = async (vault: string): Promise => { + const targets = syncTargets(getConfig()).filter((t) => t.vault === vault); + if (targets.length === 0) throw new Error(`no sync origin configured for vault '${vault}'`); + let last = busy(targets[0] as SyncTarget); + for (const t of targets) last = await runTarget(t); + return last; + }; + + const reschedule = (): void => { + for (const timer of timers.values()) clearInterval(timer); + timers.clear(); + const all = syncTargets(getConfig()); + const live = new Set(all.map(keyOf)); + for (const key of [...states.keys()]) if (!live.has(key)) states.delete(key); + for (const t of all) ensureState(t); + for (const t of autoSyncTargets(getConfig())) { + const timer = setInterval(() => void runTarget(t), intervalMs(t.intervalSeconds)); + timer.unref?.(); + timers.set(keyOf(t), timer); + } + }; + + const stop = (): void => { + for (const timer of timers.values()) clearInterval(timer); + timers.clear(); + }; + + return { reschedule, runNow, status: () => ({ vaults: [...states.values()] }), stop }; +}; diff --git a/src/sync/planner.test.ts b/src/sync/planner.test.ts new file mode 100644 index 0000000..4a56321 --- /dev/null +++ b/src/sync/planner.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it } from 'vitest'; +import { type VaultsConfig } from '@agentage/memory-core'; +import { + autoSyncTargets, + DEFAULT_IGNORE, + DEFAULT_INTERVAL_SECONDS, + intervalMs, + resolveIgnore, + syncTargets, +} from './planner.js'; + +describe('resolveIgnore', () => { + it('defaults to the editor/runtime files when absent', () => { + expect(resolveIgnore(undefined)).toEqual([...DEFAULT_IGNORE]); + }); + + it('REPLACES the defaults with a set value', () => { + expect(resolveIgnore(['secrets/'])).toEqual(['secrets/']); + }); + + it('treats an empty array as "sync everything"', () => { + expect(resolveIgnore([])).toEqual([]); + }); +}); + +describe('intervalMs', () => { + it('converts seconds to milliseconds', () => { + expect(intervalMs(300)).toBe(300_000); + }); + + it('floors and clamps negatives to zero', () => { + expect(intervalMs(1.9)).toBe(1_000); + expect(intervalMs(-5)).toBe(0); + }); +}); + +describe('syncTargets', () => { + const cfg = (vaults: NonNullable): VaultsConfig => ({ + version: 1, + vaults, + }); + + it('includes only vaults with both a path and an external origin', () => { + const targets = syncTargets( + cfg({ + synced: { path: '/tmp/synced', origin: [{ remote: 'git@h:me/s.git' }] }, + localOnly: { path: '/tmp/local' }, + remoteOnly: { origin: [{ remote: 'git@h:me/r.git' }] }, + cloud: { path: '/tmp/cloud', origin: [{ remote: 'agentage' }] }, + }) + ); + expect(targets.map((t) => t.vault)).toEqual(['synced']); + expect(targets[0]).toMatchObject({ + remote: 'git@h:me/s.git', + remoteName: 'sync', + intervalSeconds: DEFAULT_INTERVAL_SECONDS, + ignore: [...DEFAULT_IGNORE], + }); + }); + + it('flattens multiple origins into distinct remote names', () => { + const targets = syncTargets( + cfg({ + multi: { + path: '/tmp/m', + origin: [ + { remote: 'git@h:me/a.git', interval: 60, ignore: [] }, + { remote: 'git@h:me/b.git' }, + ], + }, + }) + ); + expect(targets.map((t) => t.remoteName)).toEqual(['sync', 'sync-1']); + expect(targets[0]).toMatchObject({ intervalSeconds: 60, ignore: [] }); + }); + + it('skips blank remotes', () => { + expect(syncTargets(cfg({ v: { path: '/tmp/v', origin: [{ remote: ' ' }] } }))).toEqual([]); + }); +}); + +describe('autoSyncTargets', () => { + it('excludes interval 0 (manual-only) from the daemon loop', () => { + const config: VaultsConfig = { + version: 1, + vaults: { + auto: { path: '/tmp/a', origin: [{ remote: 'git@h:me/a.git', interval: 30 }] }, + manual: { path: '/tmp/m', origin: [{ remote: 'git@h:me/m.git', interval: 0 }] }, + }, + }; + expect(autoSyncTargets(config).map((t) => t.vault)).toEqual(['auto']); + // Both are still valid sync targets; only the loop excludes the manual one. + expect( + syncTargets(config) + .map((t) => t.vault) + .sort() + ).toEqual(['auto', 'manual']); + }); +}); diff --git a/src/sync/planner.ts b/src/sync/planner.ts new file mode 100644 index 0000000..8607f01 --- /dev/null +++ b/src/sync/planner.ts @@ -0,0 +1,57 @@ +import { expandPath, type VaultsConfig } from '@agentage/memory-core'; + +// The unified schema stores interval as a bare non-negative integer; sync treats it as SECONDS +// (there is no minutes marker in the schema, so seconds is the least-surprising reading and keeps +// tests fast). 0 means manual-only (excluded from the daemon auto loop). +export const DEFAULT_INTERVAL_SECONDS = 300; + +// When `ignore` is absent these editor/runtime files are excluded from sync; a set value REPLACES +// the defaults, and an empty array syncs everything. +export const DEFAULT_IGNORE: readonly string[] = ['.obsidian/', 'data.json']; + +// The reserved cloud channel is never synced over external git (that path is out of scope here). +const RESERVED_REMOTE = 'agentage'; + +export interface SyncTarget { + vault: string; + path: string; // absolute local working-copy path + remoteName: string; // git remote name for this origin + remote: string; // remote URL + intervalSeconds: number; + ignore: string[]; +} + +export const resolveIgnore = (ignore: string[] | undefined): string[] => + ignore === undefined ? [...DEFAULT_IGNORE] : ignore; + +export const intervalMs = (seconds: number): number => Math.max(0, Math.floor(seconds)) * 1000; + +// One vault may carry several origins; each gets a distinct remote name within its own repo. +const remoteNameFor = (index: number): string => (index === 0 ? 'sync' : `sync-${index}`); + +// Flatten (vault, origin) pairs into sync targets. A target needs a local `path` (the working +// copy to commit/push from) AND an external origin; origin-only entries (cloud remote backends) +// and the reserved cloud channel are skipped. +export const syncTargets = (config: VaultsConfig): SyncTarget[] => { + const out: SyncTarget[] = []; + for (const [vault, entry] of Object.entries(config.vaults ?? {})) { + if (!entry.path || !entry.origin?.length) continue; + entry.origin.forEach((origin, index) => { + const remote = origin.remote.trim(); + if (!remote || remote === RESERVED_REMOTE) return; + out.push({ + vault, + path: expandPath(entry.path as string), + remoteName: remoteNameFor(index), + remote, + intervalSeconds: origin.interval ?? DEFAULT_INTERVAL_SECONDS, + ignore: resolveIgnore(origin.ignore), + }); + }); + } + return out; +}; + +// The targets the daemon auto-loop schedules: interval 0 is manual-only and excluded. +export const autoSyncTargets = (config: VaultsConfig): SyncTarget[] => + syncTargets(config).filter((t) => t.intervalSeconds > 0);