mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 02:06:43 +00:00
refactor: move native hook relay registry to SQLite (#109289)
* refactor: store native hook relays in SQLite * chore: satisfy relay store CI checks
This commit is contained in:
Peter Steinberger
@@ -2271,7 +2271,6 @@ Add a repo check that fails new runtime writes to legacy state paths:
|
||||
- Discord `model-picker-preferences.json`
|
||||
- Discord `command-deploy-cache.json`
|
||||
- sandbox registry shard JSON files
|
||||
- native hook relay `/tmp` bridge JSON files
|
||||
- `plugin-state/state.sqlite`
|
||||
- ad-hoc `openclaw-state.sqlite` runtime sidecars
|
||||
- `tasks/runs.sqlite`
|
||||
|
||||
@@ -108,6 +108,7 @@ const legacyStorePatterns = [
|
||||
/\bplugin-state\/state\.sqlite\b/u,
|
||||
/\btasks\/(?:runs\.sqlite|flows\/registry\.sqlite)\b/u,
|
||||
/\bopenclaw-state\.sqlite\b/u,
|
||||
/\bopenclaw-native-hook-relays\b/u,
|
||||
];
|
||||
|
||||
const allowedRuntimeMigrationPaths = [
|
||||
|
||||
@@ -16,6 +16,7 @@ const removedAsyncTransactionNames = new Set([
|
||||
]);
|
||||
const synchronousTransactionCallbackIndexes = new Map([
|
||||
["runOpenClawAgentWriteTransaction", 0],
|
||||
["runOpenClawStateWriteTransaction", 0],
|
||||
["runSqliteImmediateTransactionSync", 1],
|
||||
]);
|
||||
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
|
||||
import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js";
|
||||
import {
|
||||
deleteNativeHookRelayBridgeRecordIfOwned,
|
||||
pruneNativeHookRelayBridgeRecords,
|
||||
readNativeHookRelayBridgeRecord,
|
||||
renewOrRestoreNativeHookRelayBridgeRecord,
|
||||
type NativeHookRelayBridgeRecord,
|
||||
writeNativeHookRelayBridgeRecord,
|
||||
} from "./native-hook-relay-store.js";
|
||||
|
||||
let testRoot = "";
|
||||
let primaryStateDbPath = "";
|
||||
let secondaryStateDbPath = "";
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
beforeEach(() => {
|
||||
testRoot = tempDirs.make("openclaw-native-hook-relay-store-");
|
||||
primaryStateDbPath = path.join(testRoot, "primary.sqlite");
|
||||
secondaryStateDbPath = path.join(testRoot, "secondary.sqlite");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
});
|
||||
|
||||
function bridgeRecord(
|
||||
relayId: string,
|
||||
overrides: Partial<NativeHookRelayBridgeRecord> = {},
|
||||
): NativeHookRelayBridgeRecord {
|
||||
return {
|
||||
relayId,
|
||||
pid: 100,
|
||||
hostname: "127.0.0.1",
|
||||
port: 18_789,
|
||||
token: "test-token-placeholder",
|
||||
expiresAtMs: 20_000,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("native hook relay store", () => {
|
||||
it("upserts and reads bridge records", () => {
|
||||
const first = bridgeRecord("relay-upsert");
|
||||
const replacement = bridgeRecord("relay-upsert", {
|
||||
pid: 101,
|
||||
port: 18_790,
|
||||
token: "test-auth-token",
|
||||
expiresAtMs: 30_000,
|
||||
});
|
||||
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: first,
|
||||
updatedAtMs: 1_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: first.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(first);
|
||||
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: replacement,
|
||||
updatedAtMs: 2_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: replacement.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(replacement);
|
||||
});
|
||||
|
||||
it("requires matching token and pid to renew or delete a bridge", () => {
|
||||
const record = bridgeRecord("relay-owned");
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record,
|
||||
updatedAtMs: 1_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
|
||||
expect(
|
||||
renewOrRestoreNativeHookRelayBridgeRecord({
|
||||
record: { ...record, pid: record.pid + 1, expiresAtMs: 30_000 },
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(false);
|
||||
expect(
|
||||
renewOrRestoreNativeHookRelayBridgeRecord({
|
||||
record: { ...record, token: "decoy-token", expiresAtMs: 30_000 },
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(false);
|
||||
expect(
|
||||
renewOrRestoreNativeHookRelayBridgeRecord({
|
||||
record: { ...record, expiresAtMs: 30_000 },
|
||||
updatedAtMs: 2_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: record.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual({ ...record, expiresAtMs: 30_000 });
|
||||
|
||||
expect(
|
||||
deleteNativeHookRelayBridgeRecordIfOwned({
|
||||
...record,
|
||||
pid: record.pid + 1,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(false);
|
||||
expect(
|
||||
deleteNativeHookRelayBridgeRecordIfOwned({
|
||||
...record,
|
||||
token: "decoy-token",
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(false);
|
||||
expect(
|
||||
deleteNativeHookRelayBridgeRecordIfOwned({
|
||||
...record,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: record.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
it("restores a missing record without overwriting another owner", () => {
|
||||
const record = bridgeRecord("relay-restored");
|
||||
expect(
|
||||
renewOrRestoreNativeHookRelayBridgeRecord({
|
||||
record,
|
||||
updatedAtMs: 1_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: record.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(record);
|
||||
|
||||
const otherOwner = bridgeRecord(record.relayId, {
|
||||
pid: record.pid + 1,
|
||||
token: "test-auth-token",
|
||||
});
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: otherOwner,
|
||||
updatedAtMs: 2_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
expect(
|
||||
renewOrRestoreNativeHookRelayBridgeRecord({
|
||||
record,
|
||||
updatedAtMs: 3_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(false);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: record.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(otherOwner);
|
||||
});
|
||||
|
||||
it("does not let an old owner delete its replacement", () => {
|
||||
const oldOwner = bridgeRecord("relay-replaced", {
|
||||
pid: 100,
|
||||
token: "secret-token",
|
||||
});
|
||||
const replacement = bridgeRecord("relay-replaced", {
|
||||
pid: 101,
|
||||
port: 18_790,
|
||||
token: "test-auth-token",
|
||||
expiresAtMs: 30_000,
|
||||
});
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: oldOwner,
|
||||
updatedAtMs: 1_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: replacement,
|
||||
updatedAtMs: 2_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
|
||||
expect(
|
||||
deleteNativeHookRelayBridgeRecordIfOwned({
|
||||
...oldOwner,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBe(false);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: replacement.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(replacement);
|
||||
});
|
||||
|
||||
it("prunes expired and dead bridges while preserving live and unknown pids", () => {
|
||||
const expired = bridgeRecord("relay-expired", { pid: 200, expiresAtMs: 9_999 });
|
||||
const dead = bridgeRecord("relay-dead", { pid: 201 });
|
||||
const live = bridgeRecord("relay-live", { pid: 202 });
|
||||
const unknown = bridgeRecord("relay-unknown", { pid: 203 });
|
||||
for (const [index, record] of [expired, dead, live, unknown].entries()) {
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record,
|
||||
updatedAtMs: 1_000 + index,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
}
|
||||
const isPidDead = vi.fn((pid: number) => pid === dead.pid);
|
||||
|
||||
const pruned = pruneNativeHookRelayBridgeRecords({
|
||||
currentPid: 100,
|
||||
isPidDead,
|
||||
nowMs: 10_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
|
||||
expect(pruned).toHaveLength(2);
|
||||
expect(pruned).toEqual(
|
||||
expect.arrayContaining([
|
||||
{ relayId: expired.relayId, pid: expired.pid, reason: "expired" },
|
||||
{ relayId: dead.relayId, pid: dead.pid, reason: "dead-pid" },
|
||||
]),
|
||||
);
|
||||
expect(new Set(isPidDead.mock.calls.map(([pid]) => pid))).toStrictEqual(
|
||||
new Set([dead.pid, live.pid, unknown.pid]),
|
||||
);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: expired.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBeUndefined();
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: dead.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toBeUndefined();
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: live.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(live);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: unknown.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(unknown);
|
||||
});
|
||||
|
||||
it("preserves a replacement published during dead-pid planning", () => {
|
||||
const stale = bridgeRecord("relay-prune-race", {
|
||||
pid: 201,
|
||||
token: "secret-token",
|
||||
});
|
||||
const replacement = bridgeRecord("relay-prune-race", {
|
||||
pid: 202,
|
||||
port: 18_790,
|
||||
token: "test-auth-token",
|
||||
expiresAtMs: 30_000,
|
||||
});
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: stale,
|
||||
updatedAtMs: 1_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
|
||||
const pruned = pruneNativeHookRelayBridgeRecords({
|
||||
currentPid: 100,
|
||||
isPidDead: (pid) => {
|
||||
expect(pid).toBe(stale.pid);
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: replacement,
|
||||
updatedAtMs: 2_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
return true;
|
||||
},
|
||||
nowMs: 10_000,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
|
||||
expect(pruned).toStrictEqual([]);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: replacement.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(replacement);
|
||||
});
|
||||
|
||||
it("isolates records by the exact state database path", () => {
|
||||
const primary = bridgeRecord("relay-isolated", {
|
||||
pid: 100,
|
||||
token: "config-token",
|
||||
});
|
||||
const secondary = bridgeRecord("relay-isolated", {
|
||||
pid: 200,
|
||||
port: 18_790,
|
||||
token: "gateway-token",
|
||||
});
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: primary,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
});
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: secondary,
|
||||
stateDbPath: secondaryStateDbPath,
|
||||
});
|
||||
|
||||
expect(fs.existsSync(primaryStateDbPath)).toBe(true);
|
||||
expect(fs.existsSync(secondaryStateDbPath)).toBe(true);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: primary.relayId,
|
||||
stateDbPath: primaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(primary);
|
||||
expect(
|
||||
readNativeHookRelayBridgeRecord({
|
||||
relayId: secondary.relayId,
|
||||
stateDbPath: secondaryStateDbPath,
|
||||
}),
|
||||
).toStrictEqual(secondary);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,329 @@
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
getNodeSqliteKysely,
|
||||
} from "../../infra/kysely-sync.js";
|
||||
import { withOpenClawStateDatabaseReadOnly } from "../../state/openclaw-state-db-readonly.js";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../../state/openclaw-state-db.generated.js";
|
||||
import {
|
||||
openOpenClawStateDatabase,
|
||||
runOpenClawStateWriteTransaction,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
|
||||
export type NativeHookRelayBridgeRecord = {
|
||||
relayId: string;
|
||||
pid: number;
|
||||
hostname: "127.0.0.1";
|
||||
port: number;
|
||||
token: string;
|
||||
expiresAtMs: number;
|
||||
};
|
||||
|
||||
type NativeHookRelayBridgePruneResult = {
|
||||
relayId: string;
|
||||
pid: number;
|
||||
reason: "dead-pid" | "expired";
|
||||
};
|
||||
|
||||
type NativeHookRelayBridgeDatabase = Pick<OpenClawStateKyselyDatabase, "native_hook_relay_bridges">;
|
||||
|
||||
type NativeHookRelayBridgeRow = OpenClawStateKyselyDatabase["native_hook_relay_bridges"];
|
||||
|
||||
type NativeHookRelayBridgeSnapshot = {
|
||||
record: NativeHookRelayBridgeRecord;
|
||||
updatedAtMs: number;
|
||||
};
|
||||
|
||||
type NativeHookRelayBridgePruneCandidate = {
|
||||
snapshot: NativeHookRelayBridgeSnapshot;
|
||||
reason: NativeHookRelayBridgePruneResult["reason"];
|
||||
};
|
||||
|
||||
type NativeHookRelayBridgeStoreOptions = {
|
||||
stateDbPath?: string;
|
||||
};
|
||||
|
||||
function readNativeHookRelayBridgeSnapshot(
|
||||
row: NativeHookRelayBridgeRow | undefined,
|
||||
): NativeHookRelayBridgeSnapshot | undefined {
|
||||
if (
|
||||
!row ||
|
||||
typeof row.relay_id !== "string" ||
|
||||
row.relay_id.length === 0 ||
|
||||
!Number.isSafeInteger(row.pid) ||
|
||||
row.hostname !== "127.0.0.1" ||
|
||||
!Number.isSafeInteger(row.port) ||
|
||||
row.port <= 0 ||
|
||||
row.port > 65_535 ||
|
||||
typeof row.token !== "string" ||
|
||||
row.token.length === 0 ||
|
||||
!Number.isSafeInteger(row.expires_at_ms) ||
|
||||
!Number.isSafeInteger(row.updated_at_ms)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const { token } = row;
|
||||
return {
|
||||
record: {
|
||||
relayId: row.relay_id,
|
||||
pid: row.pid,
|
||||
hostname: row.hostname,
|
||||
port: row.port,
|
||||
token,
|
||||
expiresAtMs: row.expires_at_ms,
|
||||
},
|
||||
updatedAtMs: row.updated_at_ms,
|
||||
};
|
||||
}
|
||||
|
||||
function readNativeHookRelayBridgeSnapshotFromDatabase(params: {
|
||||
database: { db: DatabaseSync };
|
||||
relayId: string;
|
||||
}): NativeHookRelayBridgeSnapshot | undefined {
|
||||
const db = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(params.database.db);
|
||||
return readNativeHookRelayBridgeSnapshot(
|
||||
executeSqliteQueryTakeFirstSync(
|
||||
params.database.db,
|
||||
db.selectFrom("native_hook_relay_bridges").selectAll().where("relay_id", "=", params.relayId),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function sameNativeHookRelayBridgeSnapshot(
|
||||
left: NativeHookRelayBridgeSnapshot,
|
||||
right: NativeHookRelayBridgeSnapshot,
|
||||
): boolean {
|
||||
return (
|
||||
left.updatedAtMs === right.updatedAtMs &&
|
||||
left.record.relayId === right.record.relayId &&
|
||||
left.record.pid === right.record.pid &&
|
||||
left.record.hostname === right.record.hostname &&
|
||||
left.record.port === right.record.port &&
|
||||
left.record.token === right.record.token &&
|
||||
left.record.expiresAtMs === right.record.expiresAtMs
|
||||
);
|
||||
}
|
||||
|
||||
export function readNativeHookRelayBridgeRecord(
|
||||
params: { relayId: string } & NativeHookRelayBridgeStoreOptions,
|
||||
): NativeHookRelayBridgeRecord | undefined {
|
||||
return withOpenClawStateDatabaseReadOnly(
|
||||
(database) =>
|
||||
readNativeHookRelayBridgeSnapshotFromDatabase({
|
||||
database,
|
||||
relayId: params.relayId,
|
||||
})?.record,
|
||||
{ path: params.stateDbPath },
|
||||
);
|
||||
}
|
||||
|
||||
export function writeNativeHookRelayBridgeRecord(
|
||||
params: {
|
||||
record: NativeHookRelayBridgeRecord;
|
||||
updatedAtMs?: number;
|
||||
} & NativeHookRelayBridgeStoreOptions,
|
||||
): void {
|
||||
const updatedAtMs = params.updatedAtMs ?? Date.now();
|
||||
const record = params.record;
|
||||
const { token } = record;
|
||||
runOpenClawStateWriteTransaction(
|
||||
(database) => {
|
||||
const db = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(database.db);
|
||||
executeSqliteQuerySync(
|
||||
database.db,
|
||||
db
|
||||
.insertInto("native_hook_relay_bridges")
|
||||
.values({
|
||||
relay_id: record.relayId,
|
||||
pid: record.pid,
|
||||
hostname: record.hostname,
|
||||
port: record.port,
|
||||
token,
|
||||
expires_at_ms: record.expiresAtMs,
|
||||
updated_at_ms: updatedAtMs,
|
||||
})
|
||||
.onConflict((conflict) =>
|
||||
conflict.column("relay_id").doUpdateSet({
|
||||
pid: record.pid,
|
||||
hostname: record.hostname,
|
||||
port: record.port,
|
||||
token,
|
||||
expires_at_ms: record.expiresAtMs,
|
||||
updated_at_ms: updatedAtMs,
|
||||
}),
|
||||
),
|
||||
);
|
||||
},
|
||||
{ path: params.stateDbPath },
|
||||
);
|
||||
}
|
||||
|
||||
export function renewOrRestoreNativeHookRelayBridgeRecord(
|
||||
params: {
|
||||
record: NativeHookRelayBridgeRecord;
|
||||
updatedAtMs?: number;
|
||||
} & NativeHookRelayBridgeStoreOptions,
|
||||
): boolean {
|
||||
const { record } = params;
|
||||
const { token } = record;
|
||||
const updatedAtMs = params.updatedAtMs ?? Date.now();
|
||||
return runOpenClawStateWriteTransaction(
|
||||
(database) => {
|
||||
const db = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(database.db);
|
||||
const current = readNativeHookRelayBridgeSnapshotFromDatabase({
|
||||
database,
|
||||
relayId: record.relayId,
|
||||
});
|
||||
if (!current) {
|
||||
const result = executeSqliteQuerySync(
|
||||
database.db,
|
||||
db
|
||||
.insertInto("native_hook_relay_bridges")
|
||||
.values({
|
||||
relay_id: record.relayId,
|
||||
pid: record.pid,
|
||||
hostname: record.hostname,
|
||||
port: record.port,
|
||||
token,
|
||||
expires_at_ms: record.expiresAtMs,
|
||||
updated_at_ms: updatedAtMs,
|
||||
})
|
||||
.onConflict((conflict) => conflict.column("relay_id").doNothing()),
|
||||
);
|
||||
return result.numAffectedRows === 1n;
|
||||
}
|
||||
if (current.record.pid !== record.pid || current.record.token !== token) {
|
||||
return false;
|
||||
}
|
||||
const result = executeSqliteQuerySync(
|
||||
database.db,
|
||||
db
|
||||
.updateTable("native_hook_relay_bridges")
|
||||
.set({
|
||||
hostname: record.hostname,
|
||||
port: record.port,
|
||||
expires_at_ms: record.expiresAtMs,
|
||||
updated_at_ms: updatedAtMs,
|
||||
})
|
||||
.where("relay_id", "=", record.relayId)
|
||||
.where("pid", "=", record.pid)
|
||||
.where("token", "=", token)
|
||||
.where("updated_at_ms", "=", current.updatedAtMs),
|
||||
);
|
||||
return result.numAffectedRows === 1n;
|
||||
},
|
||||
{ path: params.stateDbPath },
|
||||
);
|
||||
}
|
||||
|
||||
export function deleteNativeHookRelayBridgeRecordIfOwned(params: {
|
||||
relayId: string;
|
||||
pid: number;
|
||||
token: string;
|
||||
stateDbPath?: string;
|
||||
}): boolean {
|
||||
return runOpenClawStateWriteTransaction(
|
||||
(database) => {
|
||||
const current = readNativeHookRelayBridgeSnapshotFromDatabase({
|
||||
database,
|
||||
relayId: params.relayId,
|
||||
});
|
||||
if (!current || current.record.pid !== params.pid || current.record.token !== params.token) {
|
||||
return false;
|
||||
}
|
||||
const db = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(database.db);
|
||||
const result = executeSqliteQuerySync(
|
||||
database.db,
|
||||
db
|
||||
.deleteFrom("native_hook_relay_bridges")
|
||||
.where("relay_id", "=", params.relayId)
|
||||
.where("pid", "=", params.pid)
|
||||
.where("token", "=", params.token)
|
||||
.where("updated_at_ms", "=", current.updatedAtMs),
|
||||
);
|
||||
return result.numAffectedRows === 1n;
|
||||
},
|
||||
{ path: params.stateDbPath },
|
||||
);
|
||||
}
|
||||
|
||||
export function pruneNativeHookRelayBridgeRecords(params: {
|
||||
currentPid: number;
|
||||
isPidDead: (pid: number) => boolean;
|
||||
nowMs?: number;
|
||||
stateDbPath?: string;
|
||||
}): NativeHookRelayBridgePruneResult[] {
|
||||
const nowMs = params.nowMs ?? Date.now();
|
||||
const database = openOpenClawStateDatabase({ path: params.stateDbPath });
|
||||
const db = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(database.db);
|
||||
const snapshots = executeSqliteQuerySync(
|
||||
database.db,
|
||||
db.selectFrom("native_hook_relay_bridges").selectAll(),
|
||||
).rows.flatMap((row) => {
|
||||
const snapshot = readNativeHookRelayBridgeSnapshot(row);
|
||||
return snapshot ? [snapshot] : [];
|
||||
});
|
||||
const candidates: NativeHookRelayBridgePruneCandidate[] = [];
|
||||
for (const snapshot of snapshots) {
|
||||
if (nowMs > snapshot.record.expiresAtMs) {
|
||||
candidates.push({ snapshot, reason: "expired" });
|
||||
continue;
|
||||
}
|
||||
if (snapshot.record.pid !== params.currentPid && params.isPidDead(snapshot.record.pid)) {
|
||||
candidates.push({ snapshot, reason: "dead-pid" });
|
||||
}
|
||||
}
|
||||
if (candidates.length === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
return runOpenClawStateWriteTransaction(
|
||||
(writeDatabase) => {
|
||||
const writeDb = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(writeDatabase.db);
|
||||
const pruned: NativeHookRelayBridgePruneResult[] = [];
|
||||
for (const candidate of candidates) {
|
||||
const current = readNativeHookRelayBridgeSnapshotFromDatabase({
|
||||
database: writeDatabase,
|
||||
relayId: candidate.snapshot.record.relayId,
|
||||
});
|
||||
if (
|
||||
!current ||
|
||||
!sameNativeHookRelayBridgeSnapshot(current, candidate.snapshot) ||
|
||||
(candidate.reason === "expired" && nowMs <= current.record.expiresAtMs)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const result = executeSqliteQuerySync(
|
||||
writeDatabase.db,
|
||||
writeDb
|
||||
.deleteFrom("native_hook_relay_bridges")
|
||||
.where("relay_id", "=", current.record.relayId)
|
||||
.where("token", "=", current.record.token)
|
||||
.where("updated_at_ms", "=", current.updatedAtMs),
|
||||
);
|
||||
if (result.numAffectedRows === 1n) {
|
||||
pruned.push({
|
||||
relayId: current.record.relayId,
|
||||
pid: current.record.pid,
|
||||
reason: candidate.reason,
|
||||
});
|
||||
}
|
||||
}
|
||||
return pruned;
|
||||
},
|
||||
{ path: params.stateDbPath },
|
||||
);
|
||||
}
|
||||
|
||||
export function clearNativeHookRelayBridgeRecordsForTests(
|
||||
options: NativeHookRelayBridgeStoreOptions = {},
|
||||
): void {
|
||||
runOpenClawStateWriteTransaction(
|
||||
(database) => {
|
||||
const db = getNodeSqliteKysely<NativeHookRelayBridgeDatabase>(database.db);
|
||||
executeSqliteQuerySync(database.db, db.deleteFrom("native_hook_relay_bridges"));
|
||||
},
|
||||
{ path: options.stateDbPath },
|
||||
);
|
||||
}
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
@@ -1,6 +1,5 @@
|
||||
// Covers native hook relay registration, bridge invocation, and approval state.
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { rmSync, statSync, writeFileSync } from "node:fs";
|
||||
import fs from "node:fs/promises";
|
||||
import { createServer, request as httpRequest } from "node:http";
|
||||
import { tmpdir } from "node:os";
|
||||
@@ -16,6 +15,13 @@ import { createMockPluginRegistry } from "../../plugins/hooks.test-fixtures.js";
|
||||
import { patchPluginSessionExtension } from "../../plugins/host-hook-state.js";
|
||||
import { createEmptyPluginRegistry } from "../../plugins/registry-empty.js";
|
||||
import { setActivePluginRegistry } from "../../plugins/runtime.js";
|
||||
import { resolveOpenClawStateSqlitePath } from "../../state/openclaw-state-db.paths.js";
|
||||
import {
|
||||
deleteNativeHookRelayBridgeRecordIfOwned,
|
||||
readNativeHookRelayBridgeRecord,
|
||||
writeNativeHookRelayBridgeRecord,
|
||||
type NativeHookRelayBridgeRecord,
|
||||
} from "./native-hook-relay-store.js";
|
||||
import {
|
||||
testing,
|
||||
buildNativeHookRelayCommand,
|
||||
@@ -78,13 +84,16 @@ function getOnlyNativeHookRelayInvocation() {
|
||||
|
||||
async function waitForNativeHookRelayBridgeRecord(
|
||||
relayId: string,
|
||||
): Promise<Record<string, unknown>> {
|
||||
let record: Record<string, unknown> | undefined;
|
||||
): Promise<NativeHookRelayBridgeRecord> {
|
||||
let record: NativeHookRelayBridgeRecord | undefined;
|
||||
await vi.waitFor(() => {
|
||||
record = testing.getNativeHookRelayBridgeRecordForTests(relayId);
|
||||
expect(isRecord(record) ? record.relayId : undefined).toBe(relayId);
|
||||
record = readNativeHookRelayBridgeRecord({ relayId });
|
||||
expect(record?.relayId).toBe(relayId);
|
||||
});
|
||||
return record as Record<string, unknown>;
|
||||
if (!record) {
|
||||
throw new Error(`Expected native hook relay bridge record for ${relayId}`);
|
||||
}
|
||||
return record;
|
||||
}
|
||||
|
||||
async function writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
@@ -94,33 +103,29 @@ async function writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
expiresAtMs: number;
|
||||
},
|
||||
): Promise<string> {
|
||||
// Foreign bridge records simulate another process owning the relay server,
|
||||
// without starting a second OpenClaw process in the unit test.
|
||||
const bridgeDir = testing.getNativeHookRelayBridgeDirForTests();
|
||||
await fs.mkdir(bridgeDir, { recursive: true, mode: 0o700 });
|
||||
const registryPath = testing.getNativeHookRelayBridgeRegistryPathForTests(relayId);
|
||||
writeFileSync(
|
||||
registryPath,
|
||||
`${JSON.stringify({
|
||||
version: 1,
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: {
|
||||
relayId,
|
||||
pid: record.pid,
|
||||
hostname: "127.0.0.1",
|
||||
port: 9,
|
||||
token: `token-${relayId}`,
|
||||
token: "test-token-placeholder",
|
||||
expiresAtMs: record.expiresAtMs,
|
||||
})}\n`,
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
return registryPath;
|
||||
},
|
||||
});
|
||||
return relayId;
|
||||
}
|
||||
|
||||
function uniqueNativeHookRelayIdForTests(prefix: string): string {
|
||||
return `${prefix}-${randomUUID()}`;
|
||||
}
|
||||
|
||||
function nativeHookRelayStateDbArgForTests(): string {
|
||||
return `--state-db ${resolveOpenClawStateSqlitePath()}`;
|
||||
}
|
||||
|
||||
function openDeferredNativeHookRelayBridgeRequest(
|
||||
record: Record<string, unknown>,
|
||||
record: Pick<NativeHookRelayBridgeRecord, "hostname" | "port" | "token">,
|
||||
payload: Record<string, unknown>,
|
||||
): {
|
||||
connected: Promise<void>;
|
||||
@@ -137,12 +142,12 @@ function openDeferredNativeHookRelayBridgeRequest(
|
||||
});
|
||||
const req = httpRequest(
|
||||
{
|
||||
hostname: String(record.hostname),
|
||||
hostname: record.hostname,
|
||||
method: "POST",
|
||||
path: "/invoke",
|
||||
port: Number(record.port),
|
||||
port: record.port,
|
||||
headers: {
|
||||
authorization: `Bearer ${String(record.token)}`,
|
||||
authorization: `Bearer ${record.token}`,
|
||||
"content-type": "application/json",
|
||||
"content-length": Buffer.byteLength(body),
|
||||
},
|
||||
@@ -248,15 +253,15 @@ describe("native hook relay registry", () => {
|
||||
);
|
||||
expect(relay.commandForEvent("pre_tool_use")).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
);
|
||||
expect(relay.commandForEvent("pre_tool_use", { timeoutMs: 900 })).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event pre_tool_use --timeout 900`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event pre_tool_use --timeout 900`,
|
||||
);
|
||||
expect(relay.commandForEvent("pre_tool_use", { timeoutMs: 2_000 })).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -543,7 +548,7 @@ describe("native hook relay registry", () => {
|
||||
expect(relay.shouldRelayEvent("permission_request")).toBe(true);
|
||||
expect(relay.commandForEvent("pre_tool_use")).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event pre_tool_use --pre-tool-use-unavailable noop --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event pre_tool_use --pre-tool-use-unavailable noop --timeout 1234`,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -566,7 +571,7 @@ describe("native hook relay registry", () => {
|
||||
expect(relay.shouldRelayEvent("pre_tool_use")).toBe(true);
|
||||
expect(relay.commandForEvent("pre_tool_use")).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -586,7 +591,7 @@ describe("native hook relay registry", () => {
|
||||
expect(relay.shouldRelayEvent("pre_tool_use")).toBe(true);
|
||||
expect(relay.commandForEvent("pre_tool_use")).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event pre_tool_use --timeout 1234`,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -635,7 +640,7 @@ describe("native hook relay registry", () => {
|
||||
expect(relay.shouldRelayEvent("before_agent_finalize")).toBe(false);
|
||||
expect(relay.commandForEvent("post_tool_use")).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event post_tool_use --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event post_tool_use --timeout 1234`,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -657,7 +662,7 @@ describe("native hook relay registry", () => {
|
||||
expect(relay.shouldRelayEvent("before_agent_finalize")).toBe(true);
|
||||
expect(relay.commandForEvent("before_agent_finalize")).toBe(
|
||||
"/usr/local/bin/node '/opt/Open Claw/openclaw.mjs' hooks relay --provider codex --relay-id " +
|
||||
`${relay.relayId} --generation ${relay.generation} --event before_agent_finalize --timeout 1234`,
|
||||
`${relay.relayId} ${nativeHookRelayStateDbArgForTests()} --generation ${relay.generation} --event before_agent_finalize --timeout 1234`,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -1005,8 +1010,34 @@ describe("native hook relay registry", () => {
|
||||
expect(response).toEqual({ stdout: "", stderr: "", exitCode: 0 });
|
||||
});
|
||||
|
||||
it("prunes dead foreign direct bridge registry files during registration", async () => {
|
||||
const stalePath = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
it("restores a missing direct bridge record during renewal", async () => {
|
||||
const relay = registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "codex-restored-bridge-session",
|
||||
sessionId: "session-1",
|
||||
runId: "run-1",
|
||||
allowedEvents: ["pre_tool_use"],
|
||||
ttlMs: 10_000,
|
||||
});
|
||||
const before = await waitForNativeHookRelayBridgeRecord(relay.relayId);
|
||||
expect(
|
||||
deleteNativeHookRelayBridgeRecordIfOwned({
|
||||
...before,
|
||||
stateDbPath: resolveOpenClawStateSqlitePath(),
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(testing.getNativeHookRelayBridgeRecordForTests(relay.relayId)).toBeUndefined();
|
||||
|
||||
relay.renew(20_000);
|
||||
|
||||
const after = await waitForNativeHookRelayBridgeRecord(relay.relayId);
|
||||
expect(after.port).toBe(before.port);
|
||||
expect(after.token).toBe(before.token);
|
||||
expect(after.expiresAtMs).toBeGreaterThan(before.expiresAtMs);
|
||||
});
|
||||
|
||||
it("prunes dead foreign direct bridge records during registration", async () => {
|
||||
const staleRelayId = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
uniqueNativeHookRelayIdForTests("codex-dead-foreign-bridge"),
|
||||
{
|
||||
pid: 9_999_991,
|
||||
@@ -1029,18 +1060,18 @@ describe("native hook relay registry", () => {
|
||||
});
|
||||
|
||||
expect(kill).toHaveBeenCalledWith(9_999_991, 0);
|
||||
await expect(fs.stat(stalePath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(testing.getNativeHookRelayBridgeRecordForTests(staleRelayId)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("prunes expired foreign direct bridge registry files even when their pid is alive", async () => {
|
||||
const unrelatedLivePath = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
it("prunes expired foreign direct bridge records even when their pid is alive", async () => {
|
||||
const unrelatedLiveRelayId = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
uniqueNativeHookRelayIdForTests("codex-unrelated-live-foreign-bridge"),
|
||||
{
|
||||
pid: 9_999_994,
|
||||
expiresAtMs: Date.now() + 60_000,
|
||||
},
|
||||
);
|
||||
const stalePath = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
const staleRelayId = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
uniqueNativeHookRelayIdForTests("codex-expired-foreign-bridge"),
|
||||
{
|
||||
pid: 9_999_992,
|
||||
@@ -1064,12 +1095,12 @@ describe("native hook relay registry", () => {
|
||||
|
||||
expect(kill).toHaveBeenCalledWith(9_999_994, 0);
|
||||
expect(kill).not.toHaveBeenCalledWith(9_999_992, 0);
|
||||
await expect(fs.stat(stalePath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(fs.stat(unrelatedLivePath)).resolves.toBeDefined();
|
||||
expect(testing.getNativeHookRelayBridgeRecordForTests(staleRelayId)).toBeUndefined();
|
||||
expect(testing.getNativeHookRelayBridgeRecordForTests(unrelatedLiveRelayId)).toBeDefined();
|
||||
});
|
||||
|
||||
it("preserves live unexpired foreign direct bridge registry files during registration", async () => {
|
||||
const livePath = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
it("preserves live unexpired foreign direct bridge records during registration", async () => {
|
||||
const liveRelayId = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
uniqueNativeHookRelayIdForTests("codex-live-foreign-bridge"),
|
||||
{
|
||||
pid: 9_999_993,
|
||||
@@ -1083,24 +1114,20 @@ describe("native hook relay registry", () => {
|
||||
return true;
|
||||
});
|
||||
|
||||
try {
|
||||
registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "codex-preserve-live-foreign-bridge-session",
|
||||
sessionId: "session-1",
|
||||
runId: "run-1",
|
||||
allowedEvents: ["pre_tool_use"],
|
||||
});
|
||||
registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "codex-preserve-live-foreign-bridge-session",
|
||||
sessionId: "session-1",
|
||||
runId: "run-1",
|
||||
allowedEvents: ["pre_tool_use"],
|
||||
});
|
||||
|
||||
expect(kill).toHaveBeenCalledWith(9_999_993, 0);
|
||||
await expect(fs.stat(livePath)).resolves.toBeDefined();
|
||||
} finally {
|
||||
rmSync(livePath, { force: true });
|
||||
}
|
||||
expect(kill).toHaveBeenCalledWith(9_999_993, 0);
|
||||
expect(testing.getNativeHookRelayBridgeRecordForTests(liveRelayId)).toBeDefined();
|
||||
});
|
||||
|
||||
it("preserves foreign direct bridge registry files when liveness is unknown", async () => {
|
||||
const livePath = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
it("preserves foreign direct bridge records when liveness is unknown", async () => {
|
||||
const liveRelayId = await writeForeignNativeHookRelayBridgeRecordForTests(
|
||||
uniqueNativeHookRelayIdForTests("codex-unknown-liveness-foreign-bridge"),
|
||||
{
|
||||
pid: 9_999_994,
|
||||
@@ -1114,23 +1141,19 @@ describe("native hook relay registry", () => {
|
||||
return true;
|
||||
});
|
||||
|
||||
try {
|
||||
registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "codex-preserve-unknown-liveness-foreign-bridge-session",
|
||||
sessionId: "session-1",
|
||||
runId: "run-1",
|
||||
allowedEvents: ["pre_tool_use"],
|
||||
});
|
||||
registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "codex-preserve-unknown-liveness-foreign-bridge-session",
|
||||
sessionId: "session-1",
|
||||
runId: "run-1",
|
||||
allowedEvents: ["pre_tool_use"],
|
||||
});
|
||||
|
||||
expect(kill).toHaveBeenCalledWith(9_999_994, 0);
|
||||
await expect(fs.stat(livePath)).resolves.toBeDefined();
|
||||
} finally {
|
||||
rmSync(livePath, { force: true });
|
||||
}
|
||||
expect(kill).toHaveBeenCalledWith(9_999_994, 0);
|
||||
expect(testing.getNativeHookRelayBridgeRecordForTests(liveRelayId)).toBeDefined();
|
||||
});
|
||||
|
||||
it("keeps direct bridge registry files private and loopback-only", async () => {
|
||||
it("accepts only loopback direct bridge records", async () => {
|
||||
const relay = registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "codex-private-bridge-session",
|
||||
@@ -1140,20 +1163,14 @@ describe("native hook relay registry", () => {
|
||||
});
|
||||
|
||||
const record = await waitForNativeHookRelayBridgeRecord(relay.relayId);
|
||||
const bridgeDir = testing.getNativeHookRelayBridgeDirForTests();
|
||||
const registryPath = testing.getNativeHookRelayBridgeRegistryPathForTests(relay.relayId);
|
||||
expect(statSync(bridgeDir).mode & 0o077).toBe(0);
|
||||
expect(statSync(registryPath).mode & 0o077).toBe(0);
|
||||
|
||||
writeFileSync(
|
||||
registryPath,
|
||||
`${JSON.stringify({
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
// Simulate a hostile/corrupt database row outside the typed store contract.
|
||||
record: {
|
||||
...record,
|
||||
hostname: "192.0.2.1",
|
||||
expiresAtMs: Date.now() + 10_000,
|
||||
})}\n`,
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
} as unknown as NativeHookRelayBridgeRecord,
|
||||
});
|
||||
|
||||
await expect(
|
||||
invokeNativeHookRelayBridge({
|
||||
@@ -1190,15 +1207,13 @@ describe("native hook relay registry", () => {
|
||||
|
||||
const firstRecord = await waitForNativeHookRelayBridgeRecord(first.relayId);
|
||||
await waitForNativeHookRelayBridgeRecord(second.relayId);
|
||||
writeFileSync(
|
||||
testing.getNativeHookRelayBridgeRegistryPathForTests(second.relayId),
|
||||
`${JSON.stringify({
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: {
|
||||
...firstRecord,
|
||||
relayId: second.relayId,
|
||||
expiresAtMs: Date.now() + 10_000,
|
||||
})}\n`,
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(
|
||||
invokeNativeHookRelayBridge({
|
||||
@@ -1238,16 +1253,14 @@ describe("native hook relay registry", () => {
|
||||
if (!address || typeof address === "string") {
|
||||
throw new Error("test bridge server address unavailable");
|
||||
}
|
||||
writeFileSync(
|
||||
testing.getNativeHookRelayBridgeRegistryPathForTests(relay.relayId),
|
||||
`${JSON.stringify({
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record: {
|
||||
...record,
|
||||
port: address.port,
|
||||
token: "test-token",
|
||||
expiresAtMs: Date.now() + 10_000,
|
||||
})}\n`,
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(
|
||||
invokeNativeHookRelayBridge({
|
||||
|
||||
@@ -2,15 +2,7 @@
|
||||
* Bridges native harness hook events through registered relay processes.
|
||||
*/
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import {
|
||||
chmodSync,
|
||||
existsSync,
|
||||
lstatSync,
|
||||
mkdirSync,
|
||||
readdirSync,
|
||||
readFileSync,
|
||||
rmSync,
|
||||
} from "node:fs";
|
||||
import { existsSync } from "node:fs";
|
||||
import {
|
||||
createServer,
|
||||
request as httpRequest,
|
||||
@@ -18,7 +10,6 @@ import {
|
||||
type Server,
|
||||
type ServerResponse,
|
||||
} from "node:http";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import {
|
||||
asDateTimestampMs,
|
||||
@@ -29,11 +20,11 @@ import { stripAnsi } from "../../../packages/terminal-core/src/ansi.js";
|
||||
import type { OpenClawConfig } from "../../config/types.openclaw.js";
|
||||
import { toErrorObject } from "../../infra/errors.js";
|
||||
import { resolveOpenClawPackageRootSync } from "../../infra/openclaw-root.js";
|
||||
import { privateFileStoreSync } from "../../infra/private-file-store.js";
|
||||
import { createSubsystemLogger } from "../../logging/subsystem.js";
|
||||
import { listAgentToolResultMiddlewares } from "../../plugins/agent-tool-result-middleware.js";
|
||||
import { hasGlobalHooks } from "../../plugins/hook-runner-global.js";
|
||||
import { PluginApprovalResolutions } from "../../plugins/types.js";
|
||||
import { resolveOpenClawStateSqlitePath } from "../../state/openclaw-state-db.paths.js";
|
||||
import {
|
||||
cancelDeferredPluginToolApproval,
|
||||
hasBeforeToolCallPolicy,
|
||||
@@ -49,6 +40,15 @@ import { payloadTextResult } from "../tools/common.js";
|
||||
import { callGatewayTool } from "../tools/gateway.js";
|
||||
import { runAgentHarnessAfterToolCallHook } from "./hook-helpers.js";
|
||||
import { runAgentHarnessBeforeAgentFinalizeHook } from "./lifecycle-hook-helpers.js";
|
||||
import {
|
||||
clearNativeHookRelayBridgeRecordsForTests,
|
||||
deleteNativeHookRelayBridgeRecordIfOwned,
|
||||
pruneNativeHookRelayBridgeRecords,
|
||||
readNativeHookRelayBridgeRecord as readNativeHookRelayBridgeRecordFromStore,
|
||||
renewOrRestoreNativeHookRelayBridgeRecord,
|
||||
writeNativeHookRelayBridgeRecord,
|
||||
type NativeHookRelayBridgeRecord,
|
||||
} from "./native-hook-relay-store.js";
|
||||
import { createAgentToolResultMiddlewareRunner } from "./tool-result-middleware.js";
|
||||
|
||||
type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue };
|
||||
@@ -169,6 +169,7 @@ type InvokeNativeHookRelayParams = {
|
||||
|
||||
type InvokeNativeHookRelayBridgeParams = InvokeNativeHookRelayParams & {
|
||||
registrationTimeoutMs?: number;
|
||||
stateDbPath?: string;
|
||||
timeoutMs?: number;
|
||||
};
|
||||
|
||||
@@ -332,21 +333,11 @@ type NativeHookRelayDeferredApprovalOutcome =
|
||||
|
||||
type NativeHookRelayBridgeRegistration = {
|
||||
relayId: string;
|
||||
registryPath: string;
|
||||
stateDbPath: string;
|
||||
token: string;
|
||||
server: Server;
|
||||
};
|
||||
|
||||
type NativeHookRelayBridgeRecord = {
|
||||
version: 1;
|
||||
relayId: string;
|
||||
pid: number;
|
||||
hostname: string;
|
||||
port: number;
|
||||
token: string;
|
||||
expiresAtMs: number;
|
||||
};
|
||||
|
||||
type NativeHookRelayBridgeRequestAuth = {
|
||||
provider: NativeHookRelayProvider;
|
||||
relayId: string;
|
||||
@@ -437,7 +428,10 @@ export function registerNativeHookRelay(
|
||||
throw new Error("Native hook relay expiry is outside the supported Date range");
|
||||
}
|
||||
const allowedEvents = normalizeAllowedEvents(params.allowedEvents);
|
||||
unregisterNativeHookRelay(relayId);
|
||||
const stateDbPath = resolveOpenClawStateSqlitePath();
|
||||
unregisterNativeHookRelay(relayId, undefined, {
|
||||
deferBridgeRecordRemovalMs: NATIVE_HOOK_BRIDGE_REPLACEMENT_RECORD_GRACE_MS,
|
||||
});
|
||||
const registration: ActiveNativeHookRelayRegistration = {
|
||||
relayId,
|
||||
provider: params.provider,
|
||||
@@ -459,14 +453,15 @@ export function registerNativeHookRelay(
|
||||
...(params.onPreToolUseFailure ? { onPreToolUseFailure: params.onPreToolUseFailure } : {}),
|
||||
};
|
||||
relays.set(relayId, registration);
|
||||
registerNativeHookRelayBridge(registration);
|
||||
registerNativeHookRelayBridge(registration, stateDbPath);
|
||||
const handle: ActiveNativeHookRelayRegistrationHandle = {
|
||||
...registration,
|
||||
shouldRelayEvent: (event) => nativeHookRelayEventHasLocalWork(registration, event),
|
||||
commandForEvent: (event, options) =>
|
||||
buildNativeHookRelayCommand({
|
||||
buildNativeHookRelayCommandWithStateDatabase({
|
||||
provider: params.provider,
|
||||
relayId,
|
||||
stateDbPath,
|
||||
generation: registration.generation,
|
||||
event,
|
||||
preToolUseUnavailable:
|
||||
@@ -490,12 +485,30 @@ export function registerNativeHookRelay(
|
||||
if (renewedExpiresAtMs === undefined) {
|
||||
return;
|
||||
}
|
||||
current.expiresAtMs = renewedExpiresAtMs;
|
||||
handle.expiresAtMs = renewedExpiresAtMs;
|
||||
const bridge = relayBridges.get(relayId);
|
||||
if (bridge && bridge.server.listening) {
|
||||
writeNativeHookRelayBridgeRecordForRegistration(current, bridge);
|
||||
const record = resolveNativeHookRelayBridgeRecord(current, bridge, renewedExpiresAtMs);
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
if (
|
||||
!renewOrRestoreNativeHookRelayBridgeRecord({
|
||||
record,
|
||||
stateDbPath: bridge.stateDbPath,
|
||||
})
|
||||
) {
|
||||
log.debug("native hook relay bridge record ownership changed", { relayId });
|
||||
unregisterNativeHookRelay(relayId, current);
|
||||
return;
|
||||
}
|
||||
} catch (error) {
|
||||
log.debug("failed to renew native hook relay bridge record", { error, relayId });
|
||||
return;
|
||||
}
|
||||
}
|
||||
current.expiresAtMs = renewedExpiresAtMs;
|
||||
handle.expiresAtMs = renewedExpiresAtMs;
|
||||
},
|
||||
unregister: () => unregisterNativeHookRelay(relayId, registration),
|
||||
};
|
||||
@@ -505,11 +518,12 @@ export function registerNativeHookRelay(
|
||||
function unregisterNativeHookRelay(
|
||||
relayId: string,
|
||||
expectedRegistration?: ActiveNativeHookRelayRegistration,
|
||||
options?: { deferBridgeRecordRemovalMs?: number },
|
||||
): void {
|
||||
if (expectedRegistration && relays.get(relayId) !== expectedRegistration) {
|
||||
return;
|
||||
}
|
||||
unregisterNativeHookRelayBridge(relayId);
|
||||
unregisterNativeHookRelayBridge(relayId, options);
|
||||
relays.delete(relayId);
|
||||
removeNativeHookRelayInvocations(relayId);
|
||||
removeNativeHookRelayPreToolUseApprovals(relayId);
|
||||
@@ -574,6 +588,21 @@ export function buildNativeHookRelayCommand(params: {
|
||||
executable?: string;
|
||||
nice?: number | false;
|
||||
nodeExecutable?: string;
|
||||
}): string {
|
||||
return buildNativeHookRelayCommandWithStateDatabase(params);
|
||||
}
|
||||
|
||||
function buildNativeHookRelayCommandWithStateDatabase(params: {
|
||||
provider: NativeHookRelayProvider;
|
||||
relayId: string;
|
||||
stateDbPath?: string;
|
||||
generation?: string;
|
||||
event: NativeHookRelayEvent;
|
||||
preToolUseUnavailable?: "noop";
|
||||
timeoutMs?: number;
|
||||
executable?: string;
|
||||
nice?: number | false;
|
||||
nodeExecutable?: string;
|
||||
}): string {
|
||||
const timeoutMs = normalizePositiveInteger(params.timeoutMs, DEFAULT_RELAY_TIMEOUT_MS);
|
||||
const executable = params.executable ?? resolveOpenClawCliExecutable();
|
||||
@@ -591,6 +620,7 @@ export function buildNativeHookRelayCommand(params: {
|
||||
params.provider,
|
||||
"--relay-id",
|
||||
params.relayId,
|
||||
...(params.stateDbPath ? ["--state-db", params.stateDbPath] : []),
|
||||
...(params.generation ? ["--generation", params.generation] : []),
|
||||
"--event",
|
||||
params.event,
|
||||
@@ -835,7 +865,7 @@ export async function invokeNativeHookRelayBridge(
|
||||
let lastError: unknown = new Error("native hook relay bridge not found");
|
||||
while (Date.now() - startedAt < timeoutMs) {
|
||||
try {
|
||||
const record = readNativeHookRelayBridgeRecord(relayId);
|
||||
const record = readNativeHookRelayBridgeRecord(relayId, params.stateDbPath);
|
||||
if (Date.now() > record.expiresAtMs) {
|
||||
throw new Error("native hook relay bridge expired");
|
||||
}
|
||||
@@ -1005,60 +1035,35 @@ function isNativeHookRelayBridgePidDead(pid: number): boolean {
|
||||
}
|
||||
}
|
||||
|
||||
function registerNativeHookRelayBridge(registration: ActiveNativeHookRelayRegistration): void {
|
||||
// Prune actually stale bridge files from prior gateway processes. The bridge
|
||||
// directory is scoped by OS user (uid) and is shared across all OpenClaw
|
||||
// gateways/profiles run by that user, so a record with a non-current PID is
|
||||
// NOT automatically stale — it can legitimately belong to another live
|
||||
// gateway under the same uid. Only prune records whose owning PID is dead
|
||||
// or whose expiry has passed; leave live foreign records alone.
|
||||
function registerNativeHookRelayBridge(
|
||||
registration: ActiveNativeHookRelayRegistration,
|
||||
stateDbPath: string,
|
||||
): void {
|
||||
// Liveness checks stay outside the write transaction. The store rereads each
|
||||
// authoritative row before deletion so renewal or replacement wins the race.
|
||||
try {
|
||||
const staleDir = ensureNativeHookRelayBridgeDir();
|
||||
const now = Date.now();
|
||||
for (const name of readdirSync(staleDir)) {
|
||||
if (!name.endsWith(".json")) {
|
||||
continue;
|
||||
}
|
||||
const full = path.join(staleDir, name);
|
||||
try {
|
||||
const rec = JSON.parse(readFileSync(full, "utf8")) as {
|
||||
pid?: number;
|
||||
expiresAtMs?: number;
|
||||
};
|
||||
if (!rec || typeof rec.pid !== "number" || rec.pid === process.pid) {
|
||||
continue;
|
||||
}
|
||||
const expired = typeof rec.expiresAtMs === "number" && now > rec.expiresAtMs;
|
||||
const deadPid = !expired && isNativeHookRelayBridgePidDead(rec.pid);
|
||||
if (!expired && !deadPid) {
|
||||
// Live foreign record from another same-uid gateway/profile. Preserve it.
|
||||
continue;
|
||||
}
|
||||
rmSync(full, { force: true });
|
||||
log.debug("pruned stale native hook relay bridge file", {
|
||||
file: name,
|
||||
stalePid: rec.pid,
|
||||
currentPid: process.pid,
|
||||
reason: deadPid ? "dead-pid" : "expired",
|
||||
});
|
||||
} catch {
|
||||
// ignore unparseable / racing files
|
||||
}
|
||||
const pruned = pruneNativeHookRelayBridgeRecords({
|
||||
currentPid: process.pid,
|
||||
isPidDead: isNativeHookRelayBridgePidDead,
|
||||
stateDbPath,
|
||||
});
|
||||
for (const row of pruned) {
|
||||
log.debug("pruned stale native hook relay bridge record", {
|
||||
relayId: row.relayId,
|
||||
stalePid: row.pid,
|
||||
currentPid: process.pid,
|
||||
reason: row.reason,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
log.debug("native hook relay bridge dir prune skipped", { error });
|
||||
log.debug("native hook relay bridge record prune skipped", { error });
|
||||
}
|
||||
unregisterNativeHookRelayBridge(registration.relayId, {
|
||||
deferRegistryRemovalMs: NATIVE_HOOK_BRIDGE_REPLACEMENT_RECORD_GRACE_MS,
|
||||
});
|
||||
unregisterNativeHookRelayBridge(registration.relayId);
|
||||
const token = randomUUID();
|
||||
const bridgeDir = ensureNativeHookRelayBridgeDir();
|
||||
const bridgeKey = nativeHookRelayBridgeKey(registration.relayId);
|
||||
const registryPath = path.join(bridgeDir, `${bridgeKey}.json`);
|
||||
const server = createServer();
|
||||
const bridge: NativeHookRelayBridgeRegistration = {
|
||||
relayId: registration.relayId,
|
||||
registryPath,
|
||||
stateDbPath,
|
||||
token,
|
||||
server,
|
||||
};
|
||||
@@ -1079,7 +1084,14 @@ function registerNativeHookRelayBridge(registration: ActiveNativeHookRelayRegist
|
||||
if (relayBridges.get(registration.relayId) !== bridge) {
|
||||
return;
|
||||
}
|
||||
writeNativeHookRelayBridgeRecordForRegistration(registration, bridge);
|
||||
try {
|
||||
writeNativeHookRelayBridgeRecordForRegistration(registration, bridge);
|
||||
} catch (error) {
|
||||
log.debug("failed to publish native hook relay bridge record", {
|
||||
error,
|
||||
relayId: registration.relayId,
|
||||
});
|
||||
}
|
||||
});
|
||||
server.unref();
|
||||
}
|
||||
@@ -1088,28 +1100,43 @@ function writeNativeHookRelayBridgeRecordForRegistration(
|
||||
registration: ActiveNativeHookRelayRegistration,
|
||||
bridge: NativeHookRelayBridgeRegistration,
|
||||
): void {
|
||||
const record = resolveNativeHookRelayBridgeRecord(registration, bridge);
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
writeNativeHookRelayBridgeRecord({
|
||||
record,
|
||||
stateDbPath: bridge.stateDbPath,
|
||||
});
|
||||
}
|
||||
|
||||
function resolveNativeHookRelayBridgeRecord(
|
||||
registration: ActiveNativeHookRelayRegistration,
|
||||
bridge: NativeHookRelayBridgeRegistration,
|
||||
expiresAtMs = registration.expiresAtMs,
|
||||
): NativeHookRelayBridgeRecord | undefined {
|
||||
const address = bridge.server.address();
|
||||
if (!address || typeof address === "string") {
|
||||
log.debug("native hook relay bridge server address unavailable", {
|
||||
relayId: registration.relayId,
|
||||
});
|
||||
return;
|
||||
return undefined;
|
||||
}
|
||||
const { token } = bridge;
|
||||
const record: NativeHookRelayBridgeRecord = {
|
||||
version: 1,
|
||||
relayId: registration.relayId,
|
||||
pid: process.pid,
|
||||
hostname: "127.0.0.1",
|
||||
port: address.port,
|
||||
token: bridge.token,
|
||||
expiresAtMs: registration.expiresAtMs,
|
||||
token,
|
||||
expiresAtMs,
|
||||
};
|
||||
writeNativeHookRelayBridgeRecord(bridge.registryPath, record);
|
||||
return record;
|
||||
}
|
||||
|
||||
function unregisterNativeHookRelayBridge(
|
||||
relayId: string,
|
||||
options?: { deferRegistryRemovalMs?: number },
|
||||
options?: { deferBridgeRecordRemovalMs?: number },
|
||||
): void {
|
||||
const bridge = relayBridges.get(relayId);
|
||||
if (!bridge) {
|
||||
@@ -1117,23 +1144,28 @@ function unregisterNativeHookRelayBridge(
|
||||
}
|
||||
relayBridges.delete(relayId);
|
||||
bridge.server.close();
|
||||
const record = readNativeHookRelayBridgeRecordIfExists(relayId);
|
||||
if (record?.token === bridge.token) {
|
||||
const deferRegistryRemovalMs = normalizePositiveInteger(options?.deferRegistryRemovalMs, 0);
|
||||
if (deferRegistryRemovalMs > 0) {
|
||||
// During stable-id replacement, leave the old record in place until the
|
||||
// new bridge writes over it. Hook subprocesses can then retry
|
||||
// ECONNREFUSED/stale-registration instead of observing a missing relay.
|
||||
const timeout = setTimeout(() => {
|
||||
if (readNativeHookRelayBridgeRecordIfExists(relayId)?.token === bridge.token) {
|
||||
rmSync(bridge.registryPath, { force: true });
|
||||
}
|
||||
}, deferRegistryRemovalMs);
|
||||
timeout.unref();
|
||||
return;
|
||||
const removeRecord = () => {
|
||||
try {
|
||||
deleteNativeHookRelayBridgeRecordIfOwned({
|
||||
...bridge,
|
||||
pid: process.pid,
|
||||
});
|
||||
} catch (error) {
|
||||
log.debug("failed to remove native hook relay bridge record", { error, relayId });
|
||||
}
|
||||
rmSync(bridge.registryPath, { force: true });
|
||||
};
|
||||
const deferBridgeRecordRemovalMs = normalizePositiveInteger(
|
||||
options?.deferBridgeRecordRemovalMs,
|
||||
0,
|
||||
);
|
||||
if (deferBridgeRecordRemovalMs > 0) {
|
||||
// During stable-id replacement, retain the old locator until the successor
|
||||
// upserts. The token-scoped timer cannot delete that successor.
|
||||
const timeout = setTimeout(removeRecord, deferBridgeRecordRemovalMs);
|
||||
timeout.unref();
|
||||
return;
|
||||
}
|
||||
removeRecord();
|
||||
}
|
||||
|
||||
async function handleNativeHookRelayBridgeRequest(
|
||||
@@ -1233,8 +1265,11 @@ function writeNativeHookRelayBridgeJson(
|
||||
res.end(body);
|
||||
}
|
||||
|
||||
function readNativeHookRelayBridgeRecord(relayId: string): NativeHookRelayBridgeRecord {
|
||||
const record = readNativeHookRelayBridgeRecordIfExists(relayId);
|
||||
function readNativeHookRelayBridgeRecord(
|
||||
relayId: string,
|
||||
stateDbPath?: string,
|
||||
): NativeHookRelayBridgeRecord {
|
||||
const record = readNativeHookRelayBridgeRecordIfExists(relayId, stateDbPath);
|
||||
if (!record) {
|
||||
throw new Error("native hook relay bridge not found");
|
||||
}
|
||||
@@ -1243,66 +1278,22 @@ function readNativeHookRelayBridgeRecord(relayId: string): NativeHookRelayBridge
|
||||
|
||||
function readNativeHookRelayBridgeRecordIfExists(
|
||||
relayId: string,
|
||||
stateDbPath?: string,
|
||||
): NativeHookRelayBridgeRecord | undefined {
|
||||
const registryPath = nativeHookRelayBridgeRegistryPath(relayId);
|
||||
try {
|
||||
const parsed: unknown = JSON.parse(readFileSync(registryPath, "utf8"));
|
||||
if (isNativeHookRelayBridgeRecord(parsed, relayId)) {
|
||||
return parsed;
|
||||
}
|
||||
return readNativeHookRelayBridgeRecordFromStore({ relayId, stateDbPath });
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
log.debug("failed to read native hook relay bridge registry", { error, relayId });
|
||||
}
|
||||
log.debug("failed to read native hook relay bridge record", { error, relayId });
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function isNativeHookRelayBridgeRecord(
|
||||
value: unknown,
|
||||
relayId: string,
|
||||
): value is NativeHookRelayBridgeRecord {
|
||||
return (
|
||||
isJsonObject(value) &&
|
||||
value.version === 1 &&
|
||||
value.relayId === relayId &&
|
||||
typeof value.pid === "number" &&
|
||||
Number.isInteger(value.pid) &&
|
||||
value.hostname === "127.0.0.1" &&
|
||||
typeof value.port === "number" &&
|
||||
Number.isInteger(value.port) &&
|
||||
value.port > 0 &&
|
||||
value.port <= 65_535 &&
|
||||
typeof value.token === "string" &&
|
||||
value.token.length > 0 &&
|
||||
typeof value.expiresAtMs === "number"
|
||||
);
|
||||
}
|
||||
|
||||
async function invokeNativeHookRelayBridgeRecord(params: {
|
||||
record: NativeHookRelayBridgeRecord;
|
||||
timeoutMs: number;
|
||||
payload: InvokeNativeHookRelayParams;
|
||||
}): Promise<NativeHookRelayProcessResponse> {
|
||||
const startedAt = Date.now();
|
||||
let lastError: unknown;
|
||||
while (Date.now() - startedAt < params.timeoutMs) {
|
||||
try {
|
||||
return await postNativeHookRelayBridgeRecord({
|
||||
...params,
|
||||
timeoutMs: Math.max(1, params.timeoutMs - (Date.now() - startedAt)),
|
||||
});
|
||||
} catch (error) {
|
||||
lastError = error;
|
||||
if (!isRetryableNativeHookRelayBridgeError(error)) {
|
||||
break;
|
||||
}
|
||||
await delay(
|
||||
Math.min(NATIVE_HOOK_BRIDGE_RETRY_INTERVAL_MS, params.timeoutMs - (Date.now() - startedAt)),
|
||||
);
|
||||
}
|
||||
}
|
||||
throw lastError instanceof Error ? lastError : new Error(String(lastError));
|
||||
return postNativeHookRelayBridgeRecord(params);
|
||||
}
|
||||
|
||||
function postNativeHookRelayBridgeRecord(params: {
|
||||
@@ -1401,50 +1392,6 @@ function isRetryableNativeHookRelayBridgeLookupError(params: {
|
||||
);
|
||||
}
|
||||
|
||||
function nativeHookRelayBridgeDir(): string {
|
||||
const uid = typeof process.getuid === "function" ? process.getuid() : "nouid";
|
||||
return path.join(tmpdir(), `openclaw-native-hook-relays-${uid}`);
|
||||
}
|
||||
|
||||
function ensureNativeHookRelayBridgeDir(): string {
|
||||
const bridgeDir = nativeHookRelayBridgeDir();
|
||||
mkdirSync(bridgeDir, { recursive: true, mode: 0o700 });
|
||||
const stats = lstatSync(bridgeDir);
|
||||
const expectedUid = typeof process.getuid === "function" ? process.getuid() : undefined;
|
||||
if (!stats.isDirectory() || stats.isSymbolicLink()) {
|
||||
throw new Error("unsafe native hook relay bridge directory");
|
||||
}
|
||||
if (expectedUid !== undefined && stats.uid !== expectedUid) {
|
||||
throw new Error("unsafe native hook relay bridge directory owner");
|
||||
}
|
||||
if (process.platform !== "win32" && (stats.mode & 0o077) !== 0) {
|
||||
chmodSync(bridgeDir, 0o700);
|
||||
const repaired = lstatSync(bridgeDir);
|
||||
if ((repaired.mode & 0o077) !== 0) {
|
||||
throw new Error("unsafe native hook relay bridge directory permissions");
|
||||
}
|
||||
}
|
||||
return bridgeDir;
|
||||
}
|
||||
|
||||
function writeNativeHookRelayBridgeRecord(
|
||||
registryPath: string,
|
||||
record: NativeHookRelayBridgeRecord,
|
||||
): void {
|
||||
privateFileStoreSync(path.dirname(registryPath)).writeText(
|
||||
path.basename(registryPath),
|
||||
`${JSON.stringify(record)}\n`,
|
||||
);
|
||||
}
|
||||
|
||||
function nativeHookRelayBridgeRegistryPath(relayId: string): string {
|
||||
return path.join(nativeHookRelayBridgeDir(), `${nativeHookRelayBridgeKey(relayId)}.json`);
|
||||
}
|
||||
|
||||
function nativeHookRelayBridgeKey(relayId: string): string {
|
||||
return createHash("sha256").update(relayId).digest("hex").slice(0, 32);
|
||||
}
|
||||
|
||||
function delay(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
setTimeout(resolve, Math.max(0, ms));
|
||||
@@ -2463,6 +2410,7 @@ export const testing = {
|
||||
pendingPreToolUseApprovals.clear();
|
||||
permissionApprovalWindows.clear();
|
||||
permissionAllowAlwaysApprovals.clear();
|
||||
clearNativeHookRelayBridgeRecordsForTests();
|
||||
nativeHookRelayPermissionApprovalRequester = requestNativeHookRelayPermissionApproval;
|
||||
nativeHookRelayDeferredToolApprovalRequester = requestDeferredPluginToolApproval;
|
||||
},
|
||||
@@ -2473,10 +2421,11 @@ export const testing = {
|
||||
return relays.get(relayId);
|
||||
},
|
||||
getNativeHookRelayBridgeDirForTests(): string {
|
||||
return nativeHookRelayBridgeDir();
|
||||
throw new Error("native hook relay bridge files were retired");
|
||||
},
|
||||
getNativeHookRelayBridgeRegistryPathForTests(relayId: string): string {
|
||||
return nativeHookRelayBridgeRegistryPath(relayId);
|
||||
void relayId;
|
||||
throw new Error("native hook relay bridge files were retired");
|
||||
},
|
||||
getNativeHookRelayBridgeRecordForTests(relayId: string): Record<string, unknown> | undefined {
|
||||
const record = readNativeHookRelayBridgeRecordIfExists(relayId);
|
||||
|
||||
@@ -6,6 +6,11 @@ import path from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
|
||||
import {
|
||||
registerNativeHookRelay,
|
||||
testing as nativeHookRelayTesting,
|
||||
} from "../agents/harness/native-hook-relay.js";
|
||||
import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
|
||||
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
const activeChildren = new Set<ChildProcessWithoutNullStreams>();
|
||||
@@ -13,6 +18,7 @@ const outputTimeoutMs = 20_000;
|
||||
const exitAfterOutputTimeoutMs = 5_000;
|
||||
|
||||
afterEach(async () => {
|
||||
nativeHookRelayTesting.clearNativeHookRelaysForTests();
|
||||
await Promise.all(Array.from(activeChildren, terminateChild));
|
||||
});
|
||||
|
||||
@@ -202,6 +208,51 @@ async function runHooksRelay(params: { event: "post_tool_use" | "pre_tool_use";
|
||||
}
|
||||
|
||||
describe("hooks CLI process lifecycle", () => {
|
||||
it("uses the explicit relay database when the child has a different state directory", async () => {
|
||||
const relay = registerNativeHookRelay({
|
||||
provider: "codex",
|
||||
relayId: "process-explicit-state-db",
|
||||
sessionId: "session-1",
|
||||
runId: "run-1",
|
||||
allowedEvents: ["post_tool_use"],
|
||||
});
|
||||
await expect
|
||||
.poll(() => nativeHookRelayTesting.getNativeHookRelayBridgeRecordForTests(relay.relayId))
|
||||
.toBeDefined();
|
||||
|
||||
const childStateDir = path.join(tempDirs.make("openclaw-hooks-relay-other-state-"), "state");
|
||||
await fs.mkdir(childStateDir, { recursive: true });
|
||||
const result = await runHooksCli({
|
||||
args: [
|
||||
"hooks",
|
||||
"relay",
|
||||
"--provider",
|
||||
"codex",
|
||||
"--relay-id",
|
||||
relay.relayId,
|
||||
"--state-db",
|
||||
resolveOpenClawStateSqlitePath(),
|
||||
"--generation",
|
||||
relay.generation,
|
||||
"--event",
|
||||
"post_tool_use",
|
||||
"--timeout",
|
||||
"5000",
|
||||
],
|
||||
label: "hooks relay explicit state database",
|
||||
env: {
|
||||
OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1",
|
||||
OPENCLAW_NO_RESPAWN: "1",
|
||||
OPENCLAW_STATE_DIR: childStateDir,
|
||||
},
|
||||
stdin: JSON.stringify({ hook_event_name: "PostToolUse" }),
|
||||
});
|
||||
|
||||
expect(result, result.stderr).toMatchObject({ code: 0, signal: null });
|
||||
expect(result.stderr).toBe("");
|
||||
expect(result.stdout).toBe("");
|
||||
}, 60_000);
|
||||
|
||||
it("exits after one-shot outputs when plugins leave ref'd handles", async () => {
|
||||
const fixture = await createLingeringPluginFixture();
|
||||
|
||||
|
||||
@@ -546,6 +546,7 @@ export function registerHooksCli(program: Command): void {
|
||||
.description("Internal native harness hook relay")
|
||||
.requiredOption("--provider <provider>", "Native harness provider")
|
||||
.requiredOption("--relay-id <id>", "Native hook relay id")
|
||||
.option("--state-db <path>", "Shared state database path")
|
||||
.option("--generation <generation>", "Native hook relay registration generation")
|
||||
.requiredOption("--event <event>", "Native hook event")
|
||||
.option(
|
||||
|
||||
@@ -21,6 +21,33 @@ function createWritableTextBuffer(): NodeJS.WritableStream & { text: () => strin
|
||||
}
|
||||
|
||||
describe("native hook relay CLI", () => {
|
||||
it("passes the explicit state database path to direct bridge lookup", async () => {
|
||||
const invokeBridge = vi.fn(async () => ({ stdout: "", stderr: "", exitCode: 0 }));
|
||||
|
||||
await expect(
|
||||
runNativeHookRelayCli(
|
||||
{
|
||||
provider: "codex",
|
||||
relayId: "relay-1",
|
||||
stateDb: "/tmp/profile/state/openclaw.sqlite",
|
||||
generation: "generation-1",
|
||||
event: "post_tool_use",
|
||||
},
|
||||
{
|
||||
stdin: createReadableTextStream("{}"),
|
||||
invokeBridge: invokeBridge as never,
|
||||
},
|
||||
),
|
||||
).resolves.toBe(0);
|
||||
|
||||
expect(invokeBridge).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
relayId: "relay-1",
|
||||
stateDbPath: "/tmp/profile/state/openclaw.sqlite",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("reads Codex hook JSON from stdin and forwards it to the gateway relay", async () => {
|
||||
const callGateway = vi.fn(async (_opts: unknown) => ({ stdout: "", stderr: "", exitCode: 0 }));
|
||||
const stdout = createWritableTextBuffer();
|
||||
|
||||
@@ -16,6 +16,7 @@ const MAX_NATIVE_HOOK_STDIN_BYTES = 1024 * 1024;
|
||||
export type NativeHookRelayCliOptions = {
|
||||
provider?: string;
|
||||
relayId?: string;
|
||||
stateDb?: string;
|
||||
generation?: string;
|
||||
event?: string;
|
||||
preToolUseUnavailable?: string;
|
||||
@@ -94,6 +95,7 @@ export async function runNativeHookRelayCli(
|
||||
invokeBridge({
|
||||
provider,
|
||||
relayId,
|
||||
stateDbPath: opts.stateDb?.trim() || undefined,
|
||||
generation,
|
||||
event,
|
||||
rawPayload,
|
||||
|
||||
@@ -278,6 +278,19 @@ describe("check-database-first-legacy-stores", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it("flags runtime writes to the retired native hook relay JSON registry", () => {
|
||||
const violations = collectDatabaseFirstLegacyStoreViolations(
|
||||
`
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
await fs.writeFile(path.join("/tmp", "openclaw-native-hook-relays-501", "relay.json"), "{}\n");
|
||||
`,
|
||||
"src/agents/harness/native-hook-relay-file-store.ts",
|
||||
);
|
||||
|
||||
expect(violations).toEqual([{ kind: "legacy store filesystem write", line: 4 }]);
|
||||
});
|
||||
|
||||
it("flags runtime writes to the retired subagent JSON registry", () => {
|
||||
const violations = collectDatabaseFirstLegacyStoreViolations(
|
||||
`
|
||||
|
||||
@@ -33,6 +33,7 @@ describe("SQLite transaction boundary guard", () => {
|
||||
findSqliteTransactionBoundaryViolations(`
|
||||
runSqliteImmediateTransactionSync(db, async () => await prepare());
|
||||
runOpenClawAgentWriteTransaction(async (database) => await write(database), options);
|
||||
runOpenClawStateWriteTransaction(async (database) => await write(database));
|
||||
`),
|
||||
).toEqual([
|
||||
{
|
||||
@@ -45,6 +46,11 @@ describe("SQLite transaction boundary guard", () => {
|
||||
reason:
|
||||
'passes an async callback to synchronous SQLite transaction helper "runOpenClawAgentWriteTransaction"',
|
||||
},
|
||||
{
|
||||
line: 4,
|
||||
reason:
|
||||
'passes an async callback to synchronous SQLite transaction helper "runOpenClawStateWriteTransaction"',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user