From 55fdad94d05b2a564b510aa4700a37e4585014a6 Mon Sep 17 00:00:00 2001 From: jinliyl <6469360+jinliyl@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:21:37 +0800 Subject: [PATCH] fix(dsh): preserve outbox order and redact whitespace in credentials (#657) --- integrations/dsh/README.md | 2 ++ integrations/dsh/README_ZH.md | 2 ++ integrations/dsh/src/endpoint.ts | 3 +- integrations/dsh/src/outbox.ts | 15 ++++++++- integrations/dsh/tests/client.test.mjs | 29 +++++++++++++++++ integrations/dsh/tests/dsh-tools.test.mjs | 39 +++++++++++++++++++++++ integrations/dsh/tests/outbox.test.mjs | 39 +++++++++++++++++++++++ integrations/dsh/tests/runtime.test.mjs | 34 ++++++++++++++++++++ 8 files changed, 161 insertions(+), 2 deletions(-) diff --git a/integrations/dsh/README.md b/integrations/dsh/README.md index b3f4aa28..faac78cc 100644 --- a/integrations/dsh/README.md +++ b/integrations/dsh/README.md @@ -177,6 +177,8 @@ The outbox defaults to `$DSH_HOME/reme-memory/outbox` (`~/.dsh/reme-memory/outbo Owner-lock recovery is serialized. If startup reports a busy or interrupted `.lock-recovery`, retry first; if the error persists, stop all Hosts using that queue, inspect it, and remove only the empty `.lock-recovery` directory before restarting. Preserve `.lock` and all queued JSON files. +New batches receive strictly increasing ordering timestamps, recovered from queued files after restart, so equal wall-clock times or a clock rollback cannot reorder them. Older batches that already share a timestamp retain their existing file order. + Sealed batches keep a stable `request_id` across timeouts and restarts. The updated ReMe service stores completed request receipts under its configured metadata directory, rejects reuse with different input, and replays completed results without invoking the memory agent again. An interruption before a receipt is committed can still repeat an unfinished operation; receipts do not make arbitrary agent file edits transactional. ![Automatic-memory activity](./figures/reme-status-auto-memory.png) diff --git a/integrations/dsh/README_ZH.md b/integrations/dsh/README_ZH.md index 34222a7b..e7d7cd77 100644 --- a/integrations/dsh/README_ZH.md +++ b/integrations/dsh/README_ZH.md @@ -188,6 +188,8 @@ only on retrieved memory and cite the returned paths. 进程锁的回收会串行执行。若启动报告 `.lock-recovery` 正忙或恢复中断，先重试；错误持续时，停止所有使用该队列的 Host，检查队列后仅移除空的 `.lock-recovery` 目录，再重新启动。保留 `.lock` 和所有排队的 JSON 文件。 +新批次使用严格递增的排序时间戳，重启时从排队文件恢复，避免系统时间相同或回拨导致批次乱序。已有相同时间戳的旧批次仍沿用原有文件顺序。 + 已封存的批次在超时、重启后沿用相同的 `request_id`。更新后的 ReMe 服务在配置的 metadata 目录保存已完成请求的回执，拒绝把同一个 ID 用于不同输入，并直接重放已完成结果，避免重复调用记忆 Agent。回执提交前发生中断，未完成的操作仍可能重试；回执不会把任意 Agent 文件修改变成事务。 截图测试把间隔临时设为 1；运行记录中可以看到英文测试会话形成的已完成提交： diff --git a/integrations/dsh/src/endpoint.ts b/integrations/dsh/src/endpoint.ts index b7b8f99b..d50eb5f9 100644 --- a/integrations/dsh/src/endpoint.ts +++ b/integrations/dsh/src/endpoint.ts @@ -18,5 +18,6 @@ export function validEndpoint(value: string): boolean { /** Remove URL userinfo before errors cross model, browser, or log boundaries. */ export function safeError(error: unknown): string { const message = error instanceof Error ? error.message : String(error); - return message.replace(/(https?:\/\/)[^\s/]+@/gi, "$1[redacted]@"); + // URL parsing encodes spaces and strips tabs/newlines in userinfo; do not treat them as boundaries. + return message.replace(/(https?:\/\/)[^/?#\\]+@/gi, "$1[redacted]@"); } diff --git a/integrations/dsh/src/outbox.ts b/integrations/dsh/src/outbox.ts index 3becfacd..1bc9e43c 100644 --- a/integrations/dsh/src/outbox.ts +++ b/integrations/dsh/src/outbox.ts @@ -30,6 +30,7 @@ export interface MemoryBatch { export class MemoryOutbox { private readonly lock: string; private readonly owner = `${process.pid}:${randomUUID()}`; + private latestCreatedAt = 0; constructor(readonly directory: string) { mkdirSync(directory, { recursive: true, mode: 0o700 }); @@ -68,6 +69,14 @@ export class MemoryOutbox { } finally { closeSync(fd); } + for (const name of this.files()) + this.latestCreatedAt = Math.max( + this.latestCreatedAt, + this.read(name).createdAt, + ); + } catch (error) { + this.close(); + throw error; } finally { rmdirSync(guard); } @@ -90,7 +99,11 @@ export class MemoryOutbox { : { version: 1, id: randomUUID(), - createdAt: Date.now(), + // Keep FIFO ordering when the clock repeats or moves backwards, including after restart. + createdAt: (this.latestCreatedAt = Math.max( + Date.now(), + this.latestCreatedAt + 1, + )), endpoint, sessionId, date, diff --git a/integrations/dsh/tests/client.test.mjs b/integrations/dsh/tests/client.test.mjs index af3e839b..21ed3d3c 100644 --- a/integrations/dsh/tests/client.test.mjs +++ b/integrations/dsh/tests/client.test.mjs @@ -3,6 +3,35 @@ import test from "node:test"; import { ReMeClient } from "../dist/reme/client.js"; import { safeError } from "../dist/endpoint.js"; +test("redacts URL credentials containing spaces or stripped URL whitespace", async () => { + const originalFetch = globalThis.fetch; + const client = new ReMeClient({ + endpoint: "http://localhost:2333", + requestTimeoutMs: 1000, + backgroundTimeoutMs: 1000, + }); + try { + for (const password of [ + "SECRET CANARY", + "SECRET\tCANARY", + "SECRET\nCANARY", + ]) { + const url = `http://alice:${password}@localhost:2333/search`; + assert.ok(new URL(url).password); + const expected = "refused http://[redacted]@localhost:2333/search"; + globalThis.fetch = async () => + Response.json({ success: false, answer: `refused ${url}` }); + assert.equal((await client.search("fact")).error, expected); + globalThis.fetch = async () => { + throw new Error(`refused ${url}`); + }; + assert.equal((await client.search("fact")).error, expected); + } + } finally { + globalThis.fetch = originalFetch; + } +}); + test("rejects credentials before fetch and redacts URL credentials from service and transport errors", async () => { const originalFetch = globalThis.fetch; let calls = 0; diff --git a/integrations/dsh/tests/dsh-tools.test.mjs b/integrations/dsh/tests/dsh-tools.test.mjs index a21dd4a4..6c61a706 100644 --- a/integrations/dsh/tests/dsh-tools.test.mjs +++ b/integrations/dsh/tests/dsh-tools.test.mjs @@ -1,6 +1,45 @@ import assert from "node:assert/strict"; import test from "node:test"; import { registerReMeTools } from "../dist/tools.js"; +import { ReMeClient } from "../dist/reme/client.js"; + +test("reme_search does not expose whitespace-bearing URL credentials from service errors", async () => { + const originalFetch = globalThis.fetch; + const registered = []; + globalThis.fetch = async () => + Response.json({ + success: false, + answer: "refused http://alice:SECRET CANARY@localhost:2333/search", + }); + try { + registerReMeTools( + { + tools: { + register(tool) { + registered.push(tool); + return () => {}; + }, + }, + }, + new ReMeClient({ + endpoint: "http://localhost:2333", + requestTimeoutMs: 1000, + backgroundTimeoutMs: 1000, + }), + { searchLimit: 5 }, + ); + const result = await registered[0].execute( + { query: "history" }, + { signal: new AbortController().signal }, + ); + assert.equal( + result, + "ReMe search failed: refused http://[redacted]@localhost:2333/search", + ); + } finally { + globalThis.fetch = originalFetch; + } +}); test("reme_search uses the ReMe search contract and renders model-facing text", async () => { const registered = []; diff --git a/integrations/dsh/tests/outbox.test.mjs b/integrations/dsh/tests/outbox.test.mjs index 2d102c41..9755c123 100644 --- a/integrations/dsh/tests/outbox.test.mjs +++ b/integrations/dsh/tests/outbox.test.mjs @@ -31,6 +31,41 @@ const messages = [ }, ]; +test("preserves batch order across equal timestamps, clock rollback, and restart", (t) => { + const now = Date.parse("2026-10-21T12:00:00Z"); + t.mock.timers.enable({ apis: ["Date"], now }); + const path = directory(t); + let queue = new MemoryOutbox(path); + const append = (id) => + queue.append( + "http://localhost:2333", + "session", + "2026-10-21", + [{ ...messages[0], id }], + 1, + ); + try { + append("first"); + append("second"); + const batches = readdirSync(path) + .filter((name) => name.startsWith("ready-")) + .map((name) => JSON.parse(readFileSync(join(path, name), "utf8"))); + assert.equal(new Set(batches.map((batch) => batch.createdAt)).size, 2); + queue.close(); + t.mock.timers.setTime(now - 1000); + queue = new MemoryOutbox(path); + append("third"); + const received = []; + for (let next = queue.next(); next; next = queue.next()) { + received.push(next.batch.turns[0][0].id); + queue.acknowledge(next.name); + } + assert.deepEqual(received, ["first", "second", "third"]); + } finally { + queue.close(); + } +}); + test("recovers a partial batch after the owning process exits without cleanup", (t) => { const path = directory(t); const child = spawnSync( @@ -191,6 +226,10 @@ test("preserves invalid records and keeps parser errors free of message content" (error) => !error.message.includes("PRIVATE-CONTENT"), ); assert.equal(readFileSync(join(path, name), "utf8"), "PRIVATE-CONTENT"); + queue.close(); + assert.throws(() => new MemoryOutbox(path), /Invalid ReMe outbox record/); + assert.equal(existsSync(join(path, ".lock")), false); + assert.equal(readFileSync(join(path, name), "utf8"), "PRIVATE-CONTENT"); } finally { queue.close(); } diff --git a/integrations/dsh/tests/runtime.test.mjs b/integrations/dsh/tests/runtime.test.mjs index d0ee3ff5..7cb8136f 100644 --- a/integrations/dsh/tests/runtime.test.mjs +++ b/integrations/dsh/tests/runtime.test.mjs @@ -294,6 +294,40 @@ test("does not put service reply text or file names into diagnostic snapshots", assert.doesNotMatch(JSON.stringify(runtime.snapshot()), /PRIVATE/); }); +test("keeps whitespace-bearing URL credentials out of snapshots and logs", async (t) => { + const logs = []; + const error = "refused http://alice:SECRET CANARY\n@localhost:2333"; + const runtime = makeRuntime( + t, + { + async autoMemory() { + throw new Error(error); + }, + async autoDream() { + return { ok: false, error }; + }, + }, + { ...CONFIG, autoMemoryInterval: 1 }, + { + debug() {}, + warn(...args) { + logs.push(args); + }, + }, + ); + completeTurn(runtime, { id: "credentials" }, 1, 10); + await runtime.drain(); + await runtime.runDream(); + const snapshot = runtime.snapshot(); + assert.equal(snapshot.autoMemory.recentTasks[0].phase, "failed"); + assert.equal(snapshot.autoDream.lastResult, "failed"); + assert.equal(logs.length, 2); + assert.doesNotMatch( + JSON.stringify({ snapshot, logs }), + /alice|SECRET|CANARY/, + ); +}); + test("splits dates and flushes a partial previous day without a new interval", async (t) => { const calls = []; const runtime = makeRuntime(