fix: resolve issues #135-#146 and harden tools/core

packages/core/src/runtime/history.test.ts

@@ -79,6 +79,43 @@ describe('JsonlHistorySink', () => {
         await expect(sink.flush()).resolves.toBeUndefined()
     })
 
+    test('serializes concurrent append calls without losing lines', async () => {
+        const sink = new JsonlHistorySink(filePath)
+        const events = Array.from({ length: 20 }, (_, idx) =>
+            createHistoryEvent({
+                sessionId: 'concurrent',
+                type: 'action',
+                step: idx,
+                content: `event-${idx}`,
+            }),
+        )
+
+        await Promise.all(events.map((event) => sink.append(event)))
+        await sink.flush()
+
+        const lines = (await readFile(filePath, 'utf8')).trim().split('\n').filter(Boolean)
+        expect(lines).toHaveLength(20)
+        const parsed = lines.map((line) => JSON.parse(line) as { content?: string })
+        expect(parsed[0]?.content).toBe('event-0')
+        expect(parsed[19]?.content).toBe('event-19')
+    })
+
+    test('close flushes pending writes and rejects future appends', async () => {
+        const sink = new JsonlHistorySink(filePath)
+        const first = sink.append(createHistoryEvent({ sessionId: 's-close', type: 'turn_start' }))
+        const second = sink.append(
+            createHistoryEvent({ sessionId: 's-close', type: 'assistant', content: 'done' }),
+        )
+        await sink.close()
+        await Promise.all([first, second])
+
+        const lines = (await readFile(filePath, 'utf8')).trim().split('\n').filter(Boolean)
+        expect(lines).toHaveLength(2)
+        await expect(
+            sink.append(createHistoryEvent({ sessionId: 's-close', type: 'final' })),
+        ).rejects.toThrow('History sink is closed')
+    })
+
     test('handles special characters in content', async () => {
         const sink = new JsonlHistorySink(filePath)
         const event = createHistoryEvent({

packages/core/src/runtime/history.ts

@@ -5,21 +5,40 @@ import type { HistoryEvent, HistorySink, Role } from '@memo/core/types'
 
 /** JSONL history writer: one event per line. */
 export class JsonlHistorySink implements HistorySink {
-    private ready = false
+    private ensureDirPromise: Promise<void> | null = null
+    private writeQueue: Promise<void> = Promise.resolve()
+    private closed = false
 
     constructor(private filePath: string) {}
 
+    private ensureDirectory() {
+        if (!this.ensureDirPromise) {
+            this.ensureDirPromise = mkdir(dirname(this.filePath), { recursive: true }).then(
+                () => {},
+            )
+        }
+        return this.ensureDirPromise
+    }
+
     async append(event: HistoryEvent) {
-        if (!this.ready) {
-            await mkdir(dirname(this.filePath), { recursive: true })
-            this.ready = true
+        if (this.closed) {
+            throw new Error('History sink is closed')
         }
-        await appendFile(this.filePath, `${JSON.stringify(event)}\n`, 'utf8')
+        this.writeQueue = this.writeQueue.then(async () => {
+            await this.ensureDirectory()
+            await appendFile(this.filePath, `${JSON.stringify(event)}\n`, 'utf8')
+        })
+        return this.writeQueue
     }
 
     async flush() {
-        // appendFile already ensures persistence, here only for interface compatibility
-        return Promise.resolve()
+        await this.writeQueue
+    }
+
+    async close() {
+        if (this.closed) return
+        this.closed = true
+        await this.flush()
     }
 }
 

packages/core/src/runtime/session_runtime.ts

@@ -925,12 +925,14 @@ export class AgentSessionImpl implements AgentSession {
                 },
             })
             for (const sink of this.sinks) {
-                if (sink.flush) {
-                    try {
+                try {
+                    if (sink.close) {
+                        await sink.close()
+                    } else if (sink.flush) {
                         await sink.flush()
-                    } catch (err) {
-                        console.error(`History flush failed: ${(err as Error).message}`)
                     }
+                } catch (err) {
+                    console.error(`History flush failed: ${(err as Error).message}`)
                 }
             }
         }

packages/core/src/runtime/session_runtime_helpers.test.ts

@@ -0,0 +1,79 @@
+import { describe, expect, test, vi } from 'vitest'
+import type { HistorySink } from '@memo/core/types'
+import {
+    accumulateUsage,
+    emitEventToSinks,
+    emptyUsage,
+    stableStringify,
+} from '@memo/core/runtime/session_runtime_helpers'
+
+describe('accumulateUsage', () => {
+    test('uses explicit total when provided', () => {
+        const usage = emptyUsage()
+        accumulateUsage(usage, { prompt: 2, completion: 3, total: 100 })
+        expect(usage).toEqual({ prompt: 2, completion: 3, total: 100 })
+    })
+
+    test('falls back to prompt + completion when total is absent', () => {
+        const usage = emptyUsage()
+        accumulateUsage(usage, { prompt: 2, completion: 3 })
+        expect(usage).toEqual({ prompt: 2, completion: 3, total: 5 })
+    })
+})
+
+describe('emitEventToSinks', () => {
+    test('writes structured error payload to stderr when sink append fails', async () => {
+        const writes: string[] = []
+        const writeSpy = vi.spyOn(process.stderr, 'write').mockImplementation(((chunk: unknown) => {
+            writes.push(String(chunk))
+            return true
+        }) as typeof process.stderr.write)
+
+        const failingSink: HistorySink = {
+            append: async () => {
+                throw new Error('disk full')
+            },
+        }
+
+        try {
+            await emitEventToSinks(
+                {
+                    ts: '2026-01-01T00:00:00.000Z',
+                    sessionId: 's-1',
+                    type: 'assistant',
+                    content: 'hello',
+                },
+                [failingSink],
+            )
+        } finally {
+            writeSpy.mockRestore()
+        }
+
+        expect(writes.length).toBeGreaterThan(0)
+        const parsed = JSON.parse(writes.join('').trim()) as Record<string, unknown>
+        expect(parsed.level).toBe('error')
+        expect(parsed.event).toBe('history_sink_append_failed')
+        expect(parsed.message).toBe('disk full')
+        expect(parsed.sink).toBe('Object')
+    })
+})
+
+describe('stableStringify', () => {
+    test('serializes self-referencing object without throwing', () => {
+        const root: Record<string, unknown> = {}
+        root.self = root
+
+        const serialized = stableStringify(root)
+        expect(serialized).toBe('{"self":"[Circular]"}')
+    })
+
+    test('serializes indirect circular references with circular marker', () => {
+        const parent: Record<string, unknown> = { name: 'parent' }
+        const child: Record<string, unknown> = { name: 'child', parent }
+        parent.child = child
+
+        const serialized = stableStringify(parent)
+        expect(serialized).toContain('"child":{"name":"child","parent":"[Circular]"}')
+        expect(serialized).toContain('"name":"parent"')
+    })
+})

packages/core/src/runtime/session_runtime_helpers.ts

@@ -38,6 +38,10 @@ export type ResolvedToolPermission = {
     approvalMode: 'auto' | 'strict'
 }
 
+function writeStructuredError(payload: Record<string, unknown>) {
+    process.stderr.write(`${JSON.stringify(payload)}\n`)
+}
+
 export function resolveToolPermission(options: AgentSessionOptions): ResolvedToolPermission {
     if (options.toolPermissionMode === 'none') {
         return {
@@ -122,7 +126,12 @@ export async function emitEventToSinks(event: HistoryEvent, sinks: HistorySink[]
         try {
             await sink.append(event)
         } catch (err) {
-            console.error(`Failed to write history event: ${(err as Error).message}`)
+            writeStructuredError({
+                level: 'error',
+                event: 'history_sink_append_failed',
+                sink: sink.constructor?.name || 'anonymous_sink',
+                message: (err as Error).message,
+            })
         }
     }
 }
@@ -133,14 +142,37 @@ export function isAbortError(err: unknown): err is Error {
 
 // Stable serialization for duplicate action detection (ensures consistent key ordering)
 export function stableStringify(value: unknown): string {
+    return stableStringifyWithSeen(value, new WeakSet<object>(), 0)
+}
+
+const MAX_STABLE_STRINGIFY_DEPTH = 100
+
+function stableStringifyWithSeen(value: unknown, seen: WeakSet<object>, depth: number): string {
+    if (depth > MAX_STABLE_STRINGIFY_DEPTH) {
+        return JSON.stringify('[MaxDepthExceeded]')
+    }
+    if (typeof value === 'bigint') {
+        return JSON.stringify(value.toString())
+    }
     if (value === null || typeof value !== 'object') return JSON.stringify(value) ?? 'null'
+    if (seen.has(value)) {
+        return JSON.stringify('[Circular]')
+    }
+
+    seen.add(value)
     if (Array.isArray(value)) {
-        return `[${value.map((v) => stableStringify(v)).join(',')}]`
+        const result = `[${value.map((v) => stableStringifyWithSeen(v, seen, depth + 1)).join(',')}]`
+        seen.delete(value)
+        return result
     }
     const entries = Object.entries(value as Record<string, unknown>).sort(([a], [b]) =>
         a.localeCompare(b),
     )
-    return `{${entries.map(([k, v]) => `${JSON.stringify(k)}:${stableStringify(v)}`).join(',')}}`
+    const result = `{${entries
+        .map(([k, v]) => `${JSON.stringify(k)}:${stableStringifyWithSeen(v, seen, depth + 1)}`)
+        .join(',')}}`
+    seen.delete(value)
+    return result
 }
 
 export function buildAssistantToolCalls(

packages/core/src/types.ts

@@ -371,4 +371,6 @@ export type HistorySink = {
     append: (event: HistoryEvent) => Promise<void> | void
     /** 可选：flush 持久化。 */
     flush?: () => Promise<void> | void
+    /** 可选：关闭资源并确保落盘。 */
+    close?: () => Promise<void> | void
 }

packages/tools/src/orchestrator/index.test.ts

@@ -183,4 +183,70 @@ describe('tool orchestrator', () => {
             }
         }
     })
+
+    test('rejects oversized raw string input before json parsing', async () => {
+        let executed = false
+        const orchestrator = createToolOrchestrator({
+            tools: {
+                read_file: {
+                    name: 'read_file',
+                    execute: async () => {
+                        executed = true
+                        return { content: [{ type: 'text', text: 'ok' }] }
+                    },
+                },
+            },
+        })
+
+        const result = await orchestrator.executeAction({
+            name: 'read_file',
+            input: 'x'.repeat(100_001),
+        })
+
+        assert.strictEqual(result.success, false)
+        assert.strictEqual(result.status, 'input_invalid')
+        assert.ok(result.observation.includes('input string too large'))
+        assert.strictEqual(executed, false)
+    })
+
+    test('rejects non-object json input payloads', async () => {
+        const orchestrator = createToolOrchestrator({
+            tools: {
+                read_file: {
+                    name: 'read_file',
+                    execute: async () => ({ content: [{ type: 'text', text: 'ok' }] }),
+                },
+            },
+        })
+
+        const result = await orchestrator.executeAction({
+            name: 'read_file',
+            input: '[]',
+        })
+
+        assert.strictEqual(result.success, false)
+        assert.strictEqual(result.status, 'input_invalid')
+        assert.ok(result.observation.includes('expected object'))
+    })
+
+    test('rejects validateInput success payload when shape is not object', async () => {
+        const orchestrator = createToolOrchestrator({
+            tools: {
+                read_file: {
+                    name: 'read_file',
+                    validateInput: () => ({ ok: true, data: 'not-object' }),
+                    execute: async () => ({ content: [{ type: 'text', text: 'ok' }] }),
+                },
+            },
+        })
+
+        const result = await orchestrator.executeAction({
+            name: 'read_file',
+            input: { file_path: '/tmp/a' },
+        })
+
+        assert.strictEqual(result.success, false)
+        assert.strictEqual(result.status, 'input_invalid')
+        assert.ok(result.observation.includes('expected object'))
+    })
 })

packages/tools/src/orchestrator/index.ts

@@ -13,6 +13,7 @@ import type {
 } from './types'
 
 const DEFAULT_MAX_TOOL_RESULT_CHARS = 12_000
+const MAX_TOOL_INPUT_STRING_CHARS = 100_000
 
 function getMaxToolResultChars() {
     const raw = process.env.MEMO_TOOL_RESULT_MAX_CHARS?.trim()
@@ -76,9 +77,23 @@ function flattenCallToolResult(result: CallToolResult): string {
     return texts.join('\n')
 }
 
-function parseToolInput(tool: OrchestratorTool, rawInput: unknown) {
+type ParseToolInputResult =
+    | { ok: true; data: Record<string, unknown> }
+    | { ok: false; error: string }
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+    return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+function parseToolInput(tool: OrchestratorTool, rawInput: unknown): ParseToolInputResult {
     let candidate: unknown = rawInput
     if (typeof rawInput === 'string') {
+        if (rawInput.length > MAX_TOOL_INPUT_STRING_CHARS) {
+            return {
+                ok: false as const,
+                error: `${tool.name} invalid input: input string too large (max ${MAX_TOOL_INPUT_STRING_CHARS} chars)`,
+            }
+        }
         const trimmed = rawInput.trim()
         if (trimmed) {
             try {
@@ -91,12 +106,17 @@ function parseToolInput(tool: OrchestratorTool, rawInput: unknown) {
         }
     }
 
-    if (typeof candidate !== 'object' || candidate === null) {
+    if (!isRecord(candidate)) {
         return { ok: false as const, error: `${tool.name} invalid input: expected object` }
     }
 
     if (typeof tool.validateInput === 'function') {
-        return tool.validateInput(candidate)
+        const validated = tool.validateInput(candidate)
+        if (!validated.ok) return validated
+        if (!isRecord(validated.data)) {
+            return { ok: false as const, error: `${tool.name} invalid input: expected object` }
+        }
+        return { ok: true as const, data: validated.data }
     }
 
     return { ok: true as const, data: candidate }