Skip to content
Merged
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,9 @@ import { GitStorage } from '@git-stunts/git-warp/storage';
const storage = await GitStorage.open({ cwd: '.' });
const warp = await openWarp({ storage, writer: 'agent-1' });
const team = await warp.timeline('team');

// After the final lane operation:
await storage.close();
```

Application code writes with `timeline.write(intent)` and reads with
Expand Down
11 changes: 9 additions & 2 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Added `Timeline.draft(name)`, draft writes, `Timeline.previewJoin(draft)`,
and `Timeline.join(draft)` with join receipts for first-use speculative
workflows.
- Added idempotent `GitStorage.close()` and async-disposal support to release
local Git and git-cas processes without changing history or retention.
- Added the `lint:test-law` gate to reject conditional bare `return;`
statements in test bodies so skipped assertions cannot masquerade as passing
tests.
Expand Down Expand Up @@ -50,8 +52,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Raised the coverage ratchet from `92.10%` to `92.62%` after adding targeted
coverage for bounded query node paging and memory-budget rejection paths and
removing retired compatibility surfaces.
- Upgraded `@git-stunts/git-cas` to `^6.5.0` so Git-backed materializations can
use managed `CacheSet` retention and opaque page and bundle handles.
- Upgraded `@git-stunts/git-cas` to `^6.5.3` so Git-backed materializations can
use managed `CacheSet` retention, opaque page and bundle handles, and
persistent Git object sessions.
- Moved shadow-trie leaf and branch storage from unretained raw Git blobs and
trees to bounded git-cas pages and composable bundle handles. Production
storage now shares one git-cas facade with the trie adapter, and the storage
Expand All @@ -72,6 +75,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

### Fixed

- Reused repository-scoped Git object sessions across immutable history and
CAS operations, then closed history, materialization, and CAS owners at
application, CLI, migration, and test boundaries. This removes repeated
`cat-file` and `mktree` process startup without leaking subprocesses.
- Live exact node-property reads now retain collision-safe property shards as
an independently addressed materialization root. Each node uses its full
BLAKE3 route key and a deterministic schema-v2 entry-bag envelope; writes and
Expand Down
5 changes: 5 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -104,8 +104,13 @@ if (role.receipt.outcome === 'accepted') {
} else {
console.error(role.receipt.reason, role.receipt.repairHints);
}

await storage.close();
```

`GitStorage.close()` releases local Git and git-cas processes only. It does not
delete timelines, rewrite history, or change retention anchors.

The v18 graph-first API is not exported in v19. Migrate uses of
`openWarpWorldline()`, `openWarpGraph()`, `WarpApp`, `WarpCore`,
`GitGraphAdapter`, patch builders, and graph operation creators before
Expand Down
72 changes: 54 additions & 18 deletions bin/cli/commands/mcp.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
import fs from 'node:fs';
import process from 'node:process';
import readline from 'node:readline';

import { usageError } from '../infrastructure.ts';
import { openGraph } from '../shared.ts';
import { openGraph, readCliPackageVersion } from '../shared.ts';
import {
handleMcpMessage,
mcpParseError,
Expand All @@ -14,17 +13,9 @@ import type { CliOptions } from '../types.ts';
type McpCommandResult = {
readonly payload: undefined;
readonly close: () => Promise<void>;
readonly completion: Promise<void>;
};

function readPackageVersion(): string {
const packageUrl = new URL('../../../package.json', import.meta.url);
const packageText = fs.readFileSync(packageUrl, 'utf8');
const packageJson = JSON.parse(packageText) as { readonly version?: string };
return typeof packageJson.version === 'string' && packageJson.version.length > 0
? packageJson.version
: '0.0.0';
}

function writeResponse(response: McpResponse): void {
process.stdout.write(`${JSON.stringify(response)}\n`);
}
Expand All @@ -42,25 +33,70 @@ export default async function handleMcp({
}

const { graph } = await openGraph(options);
const serverVersion = readPackageVersion();
return createMcpCommandResult(graph, readCliPackageVersion());
}

function createMcpCommandResult(
graph: Parameters<typeof handleMcpMessage>[0],
serverVersion: string,
): McpCommandResult {
const lines = readline.createInterface({
input: process.stdin,
terminal: false,
});

lines.on('line', (line) => {
void dispatchLine(graph, serverVersion, line);
});
const completion = trackMcpLines(lines, graph, serverVersion);

return {
payload: undefined,
close: () => {
completion,
close: async () => {
lines.close();
return Promise.resolve();
await completion;
},
};
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

function trackMcpLines(
lines: readline.Interface,
graph: Parameters<typeof handleMcpMessage>[0],
serverVersion: string,
): Promise<void> {
const pending = new Set<Promise<void>>();
const completedFailures: unknown[] = [];
const completion = Promise.withResolvers<void>();

lines.on('line', (line) => {
const operation = dispatchLine(graph, serverVersion, line);
pending.add(operation);
void operation.then(
() => pending.delete(operation),
(error: unknown) => {
pending.delete(operation);
completedFailures.push(error);
},
);
});
lines.once('close', () => {
void settlePendingDispatches(pending, completedFailures)
.then(completion.resolve, completion.reject);
});
return completion.promise;
}

async function settlePendingDispatches(
pending: ReadonlySet<Promise<void>>,
completedFailures: readonly unknown[],
): Promise<void> {
const failures = [...completedFailures];
const results = await Promise.allSettled([...pending]);
failures.push(...results
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
.map((result) => result.reason as unknown));
if (failures.length > 0) {
throw new AggregateError(failures, 'MCP requests failed while stdin was closing');
}
}

async function dispatchLine(
graph: Parameters<typeof handleMcpMessage>[0],
serverVersion: string,
Expand Down
22 changes: 22 additions & 0 deletions bin/cli/lifecycle.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
type CloseResource = () => Promise<void>;

/** Drains a long-running command before releasing the storage it may still use. */
export async function closeCommandResources(
closeCommand: CloseResource,
closeStorage: CloseResource,
): Promise<void> {
const failures: unknown[] = [];
for (const close of [closeCommand, closeStorage]) {
try {
await close();
} catch (error) {
failures.push(error);
}
}
if (failures.length === 1) {
throw failures[0];
}
if (failures.length > 1) {
throw new AggregateError(failures, 'CLI command and storage failed to close cleanly');
}
}
59 changes: 47 additions & 12 deletions bin/cli/shared.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,23 +30,51 @@ export type CliStorageBinding = {
readonly hookPaths: HookPathPort;
};

const activeCliStorages = new Set<GitStorage>();

/** Releases every storage composition opened by the current CLI invocation. */
export async function closeCliStorages(): Promise<void> {
const storages = [...activeCliStorages];
activeCliStorages.clear();
const results = await Promise.allSettled(storages.map(async (storage) => await storage.close()));
const failures = results
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
.map((result) => result.reason as unknown);
if (failures.length > 0) {
throw new AggregateError(failures, 'CLI storage failed to close cleanly');
}
}

/**
* Creates a persistence adapter for the given repository path.
*/
export async function createPersistence(repoPath: string): Promise<CliStorageBinding> {
const storage = await GitStorage.open({ cwd: repoPath });
const binding = resolveWarpStorage(storage);
if (!(binding.history instanceof GitTimelineHistoryAdapter)
|| binding.createTrustChain === undefined
|| binding.hookPaths === undefined) {
throw usageError('GitStorage returned an incomplete CLI storage binding');
try {
const binding = resolveWarpStorage(storage);
if (!(binding.history instanceof GitTimelineHistoryAdapter)
|| binding.createTrustChain === undefined
|| binding.hookPaths === undefined) {
throw usageError('GitStorage returned an incomplete CLI storage binding');
}
activeCliStorages.add(storage);
return {
persistence: binding.history,
runtimeStorage: binding.runtimeStorage,
createTrustChain: binding.createTrustChain,
hookPaths: binding.hookPaths,
};
} catch (error) {
try {
await storage.close();
} catch (closeError) {
throw new AggregateError(
[error, closeError],
'CLI storage binding failed and local resources did not close cleanly',
);
}
throw error;
}
return {
persistence: binding.history,
runtimeStorage: binding.runtimeStorage,
createTrustChain: binding.createTrustChain,
hookPaths: binding.hookPaths,
};
}

/**
Expand Down Expand Up @@ -188,6 +216,13 @@ export function createHookInstaller(hookPathPort: HookPathPort): HookInstaller {
});
}

/** Reads the package version from either source or built CLI layouts. */
export function readCliPackageVersion(): string {
const packageRoot = findPackageRoot(fileURLToPath(new URL('.', import.meta.url)));
const rawJson = fs.readFileSync(path.join(packageRoot, 'package.json'), 'utf8');
return readPackageVersion(rawJson);
}

/**
* Finds the repository/package root from either source or built CLI paths.
*/
Expand All @@ -199,7 +234,7 @@ function findPackageRoot(startDir: string): string {
}
const parent = path.dirname(current);
if (parent === current) {
throw usageError('Unable to locate package.json for hook installation');
throw usageError('Unable to locate the git-warp package root');
}
current = parent;
}
Expand Down
76 changes: 62 additions & 14 deletions bin/warp-graph.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@ import { installDefaultRuntimeHostNodePorts } from '../src/application/RuntimeHo
import { EXIT_CODES, HELP_TEXT, CliError, parseArgs, usageError } from './cli/infrastructure.ts';
import { stableStringify, compactStringify } from './presenters/json.ts';
import { COMMANDS } from './cli/commands/registry.ts';
import { closeCliStorages } from './cli/shared.ts';
import { closeCommandResources } from './cli/lifecycle.ts';

installDefaultRuntimeHostNodePorts();

Expand All @@ -26,12 +28,22 @@ function hasPayload(value: unknown): value is { payload: unknown; exitCode?: num
}

/** Runtime guard: does this value carry an async `close` function? */
function hasCloseFn(value: unknown): value is { close: () => Promise<void> } {
function hasCloseFn(value: unknown): value is {
close: () => Promise<void>;
completion?: Promise<void>;
} {
if (typeof value !== 'object' || value === null) { return false; }
const rec = value as Record<string, unknown>;
return typeof rec['close'] === 'function';
}

function hasCompletion(value: { readonly completion?: Promise<void> }): value is {
readonly completion: Promise<void>;
} {
return value.completion !== undefined
&& typeof value.completion.then === 'function';
}

/** Normalizes any handler return shape into { payload, exitCode }. */
function normalizeResult(result: unknown): NormalizedCommandResult {
if (hasPayload(result)) {
Expand Down Expand Up @@ -71,16 +83,41 @@ function handleEarlyExits(parsed: ParsedInvocation): boolean {

/** Registers SIGINT/SIGTERM handlers that shut down a long-running
* command gracefully. */
function installShutdownHandlers(close: () => Promise<void>): void {
let closing = false;
const shutdown = async (): Promise<void> => {
if (closing) { return; }
closing = true;
await close();
process.exit(EXIT_CODES.OK);
function installShutdownHandlers(close: () => Promise<void>): () => Promise<void> {
let shutdownPromise: Promise<void> | null = null;
const shutdown = (): Promise<void> => {
shutdownPromise ??= closeCommandResources(close, closeCliStorages);
return shutdownPromise;
};
const exitAfterShutdown = (): void => {
void shutdown().then(
() => process.exit(EXIT_CODES.OK),
() => process.exit(EXIT_CODES.INTERNAL),
);
};
process.on('SIGINT', () => { shutdown().catch(() => process.exit(1)); });
process.on('SIGTERM', () => { shutdown().catch(() => process.exit(1)); });
process.once('SIGINT', exitAfterShutdown);
process.once('SIGTERM', exitAfterShutdown);
return shutdown;
}

async function finishCompletedCommand(
completion: Promise<void>,
shutdown: () => Promise<void>,
): Promise<void> {
const [operation] = await Promise.allSettled([completion]);
const [cleanup] = await Promise.allSettled([shutdown()]);
if (operation.status === 'rejected' && cleanup.status === 'rejected') {
throw new AggregateError(
[operation.reason, cleanup.reason],
'CLI command completion and shutdown both failed',
);
}
if (operation.status === 'rejected') {
throw operation.reason;
}
if (cleanup.status === 'rejected') {
throw cleanup.reason;
}
}

/** Writes the payload (when present) in the requested stringify
Expand Down Expand Up @@ -115,16 +152,27 @@ async function main(): Promise<void> {
emitPayload(normalized.payload, options.ndjson);

// Long-running commands may return a `close` function.
// Wait for SIGINT/SIGTERM instead of exiting immediately.
// Wait for normal completion or SIGINT/SIGTERM instead of exiting immediately.
if (hasCloseFn(result)) {
installShutdownHandlers(result.close);
return; // Keep the process alive
const shutdown = installShutdownHandlers(result.close);
if (hasCompletion(result)) {
await finishCompletedCommand(result.completion, shutdown);
process.exit(normalized.exitCode);
}
return;
}

await closeCliStorages();
process.exit(normalized.exitCode);
}

main().catch((error: unknown) => {
main().catch(async (caught: unknown) => {
let error = caught;
try {
await closeCliStorages();
} catch (closeError) {
error = new AggregateError([caught, closeError], 'CLI command and storage cleanup failed');
}
const exitCode = error instanceof CliError ? error.exitCode : EXIT_CODES.INTERNAL;
const code = error instanceof CliError ? error.code : 'E_INTERNAL';
const message = error instanceof Error ? error.message : 'Unknown error';
Expand Down
4 changes: 3 additions & 1 deletion docs/TECHNICAL_TEARDOWN.md
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,9 @@ flowchart TB
```

The root import has no storage bootstrap side effect. `GitStorage.open()` owns
the Node Git and git-cas composition used by first-use applications.
the Node Git and git-cas composition used by first-use applications;
`GitStorage.close()` releases that composition's local processes without
changing history or retention.

## System Architecture

Expand Down
Loading
Loading