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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 101 additions & 0 deletions packages/sdk/src/bootstrap-image-resizer.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
import { afterEach, describe, expect, it, spyOn } from 'bun:test'
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { bootstrap } from './bootstrap.js'
import { createBunPlatform } from './bun-platform/index.js'
import type { Config } from './config.js'
import { SessionFileStore } from './core/file-store/file-store.js'
import type { ImageResizeOptions, ImageResizer } from './core/image/types.js'
import { ModelId } from './core/llm/schema.js'
import type { Platform, ProcessRunner } from './platform/index.js'
import { createTestPreset } from './testing/preset-helpers.js'

const cleanups: Array<() => Promise<void> | void> = []
afterEach(async () => {
for (const cleanup of cleanups.splice(0).reverse()) await cleanup()
})

function recordingProcess() {
const commands: string[] = []
const fail = (command: string): never => {
commands.push(command)
throw new Error('this host has no process table')
}
const process: ProcessRunner = { spawn: fail, execFile: fail }
return { process, commands }
}

/** Boots with a real Anthropic provider whose HTTP calls are answered locally. */
async function boot(platform: Platform) {
const rejectRequest = async () => Response.json({ type: 'error', error: { type: 'invalid_request_error', message: 'test' } }, { status: 400 })
const fetchSpy = spyOn(globalThis, 'fetch').mockImplementation(Object.assign(rejectRequest, { preconnect: globalThis.fetch.preconnect }))
cleanups.push(() => fetchSpy.mockRestore())
const dataPath = await mkdtemp(join(tmpdir(), 'roj-bootstrap-image-resizer-'))
cleanups.push(() => rm(dataPath, { recursive: true, force: true }))
const config: Config = {
port: 0,
host: 'localhost',
dataPath,
persistence: 'memory',
logLevel: 'error',
logFormat: 'console',
anthropicApiKey: 'test-key',
llmLoggingEnabled: false,
}
const services = bootstrap(config, { presets: [createTestPreset()] }, platform)
const imagePath = join(dataPath, 'screenshot.png')
await writeFile(imagePath, Buffer.from([0x89, 0x50, 0x4e, 0x47]))
return { services, dataPath, imagePath }
}

async function sendImageToModel(host: Awaited<ReturnType<typeof boot>>) {
const anthropic = host.services.llmProviders.get('anthropic')
if (!anthropic) throw new Error('Anthropic provider was not registered')
await anthropic.inference({
model: ModelId('claude-test'),
systemPrompt: '',
messages: [{ role: 'user', content: [{ type: 'image_url', imageUrl: { url: `file://${host.imagePath}` } }] }],
})
}

async function classifyImage(host: Awaited<ReturnType<typeof boot>>) {
const classifier = host.services.preprocessorRegistry.getForMimeType('image/png')
if (!classifier) throw new Error('No image preprocessor registered')
const files = new SessionFileStore(host.dataPath, undefined, false, host.services.platform.fs, 'session')
await classifier.process(host.imagePath, 'image/png', { files: files.scoped('upload') })
}

describe('bootstrap image resizer', () => {
it('uses the host resizer for model images and upload classification', async () => {
const trap = recordingProcess()
const resizeCalls: Array<{ filePath: string; options?: ImageResizeOptions }> = []
const imageResizer: ImageResizer = {
async resize(filePath, mimeType, options) {
resizeCalls.push({ filePath, options })
return { path: filePath, mimeType }
},
}
const host = await boot({ ...createBunPlatform(), process: trap.process, imageResizer })

await sendImageToModel(host)
await classifyImage(host)

expect(resizeCalls).toEqual([
{ filePath: host.imagePath, options: { maxFileSizeBytes: 5 * 1024 * 1024 } },
{ filePath: host.imagePath, options: expect.objectContaining({ maxDimension: 1024, maxFileSizeBytes: 1024 * 1024 }) },
])
expect(trap.commands).toEqual([])
})

it('resizes with vips when the host provides no resizer', async () => {
const trap = recordingProcess()
const host = await boot({ ...createBunPlatform(), process: trap.process })

await sendImageToModel(host)
expect(trap.commands).toEqual(['vipsheader'])

await classifyImage(host)
expect(trap.commands).toEqual(['vipsheader', 'vipsheader'])
})
})
15 changes: 6 additions & 9 deletions packages/sdk/src/bootstrap.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import type { EventStore } from '~/core/events/event-store.js'
import { FileEventStore, MemoryEventStore } from '~/core/events/index.js'
import type { FileStore } from '~/core/file-store/types.js'
import { DefaultImageProcessor } from '~/core/image/image-processor.js'
import type { ImageResizer } from '~/core/image/types.js'
import { VipsImageResizer } from '~/core/image/vips-resizer.js'
import { AnthropicProvider } from '~/core/llm/anthropic.js'
import { LLMLogger, LoggingLLMProvider, MockLLMProvider } from '~/core/llm/index.js'
Expand Down Expand Up @@ -173,7 +174,8 @@ export function bootstrap(config: Config, userConfig: RojConfig, platform: Platf
? new MemoryEventStore()
: new FileEventStore(config.dataPath, platform.fs, logger))

const { llmProvider, llmProviders, llmLogger } = createLLMProvider(config, logger, platform)
const imageResizer = platform.imageResizer ?? new VipsImageResizer({ fs: platform.fs, process: platform.process, tmpDir: platform.tmpDir })
const { llmProvider, llmProviders, llmLogger } = createLLMProvider(config, logger, platform, imageResizer)

// The one place the declared sandbox settings are folded in — every host lands here.
const resolvedPresets = applySandboxSettings(userConfig)
Expand All @@ -188,12 +190,8 @@ export function bootstrap(config: Config, userConfig: RojConfig, platform: Platf

const preprocessorRegistry = new PreprocessorRegistry()
const imageClassifierGate = new Semaphore(config.imageClassifierConcurrency ?? 10)
// Dedicated resizer for the classification path: separate from the LLM
// provider's general-purpose ImageProcessor (which keeps a higher
// maxDimension for agent file-inspection tool calls). The classifier
// hands each image to the resizer with its own 1024px override.
const classifierImageResizer = new VipsImageResizer({ fs: platform.fs, process: platform.process, tmpDir: platform.tmpDir })
preprocessorRegistry.register(new ImageClassifierPreprocessor({ llmProvider, logger, fs: platform.fs, gate: imageClassifierGate, imageResizer: classifierImageResizer }))
// Shares the LLM provider's resizer; the classifier asks it for 1024px images per call.
preprocessorRegistry.register(new ImageClassifierPreprocessor({ llmProvider, logger, fs: platform.fs, gate: imageClassifierGate, imageResizer }))
// PdfPreprocessor must come before MarkitdownPreprocessor — both could
// match `application/pdf` in principle, but the registry uses first-hit
// and PdfPreprocessor's `pdftotext + pdfimages -all + streaming` pipeline
Expand Down Expand Up @@ -228,7 +226,7 @@ function createLogger(config: Config): Logger {
: new ConsoleLogger({ level: config.logLevel })
}

function createLLMProvider(config: Config, logger: Logger, platform: Platform): {
function createLLMProvider(config: Config, logger: Logger, platform: Platform, imageResizer: ImageResizer): {
llmProvider: LLMProvider
llmProviders: ReadonlyMap<string, LLMProvider>
llmLogger?: LLMLogger
Expand All @@ -238,7 +236,6 @@ function createLLMProvider(config: Config, logger: Logger, platform: Platform):
return { llmProvider: mock, llmProviders: new Map([['mock', mock]]) }
}

const imageResizer = new VipsImageResizer({ fs: platform.fs, process: platform.process, tmpDir: platform.tmpDir })
const imageProcessor = new DefaultImageProcessor(imageResizer, platform.fs)

const routableProviders: RoutableLLMProvider[] = []
Expand Down
6 changes: 6 additions & 0 deletions packages/sdk/src/core/image/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,19 @@ export interface ImageResizeOptions {
signal?: AbortSignal
}

/**
* Paths are on the host's `FileSystem`: the caller reads the result through it and unlinks
* `tempFile` there once the image is encoded. When an image cannot be shrunk, return the original
* path: the LLM path treats a throw as an unreadable image.
*/
export interface ImageResizer {
resize(filePath: string, mimeType: string, options?: ImageResizeOptions): Promise<ImageResizeResult>
}

export interface ImageResizeResult {
path: string
mimeType: string
/** A file the resizer created; the caller owns it from here and deletes it. */
tempFile?: string
}

Expand Down
3 changes: 3 additions & 0 deletions packages/sdk/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,9 @@ export type {
ExecFileResult,
FileHandle,
FileSystem,
ImageResizeOptions,
ImageResizer,
ImageResizeResult,
Platform,
ProcessRunner,
SpawnOptions,
Expand Down
7 changes: 7 additions & 0 deletions packages/sdk/src/platform/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
* runtime packages (e.g. `@roj-ai/sdk/bun-platform`) and wired in at bootstrap time.
*/

import type { ImageResizer } from '../core/image/types.js'
import type { FileSystem } from './fs.js'
import type { FsRevision } from './fs-revision.js'
import type { GitClient } from './git.js'
Expand Down Expand Up @@ -39,6 +40,7 @@ export type { LiveScheduler, Scheduler, WakeHandler } from './scheduler.js'
export type { SessionLogPage, SessionLogStore } from './session-log.js'
export type { ShellConfinement, ShellGrant, ShellLimits, ShellRunner, ShellRunOptions, ShellRunResult } from './shell.js'

export type { ImageResizeOptions, ImageResizer, ImageResizeResult } from '../core/image/types.js'
export type { ServiceExecutorPort } from '../plugins/services/service.js'
export type { PortPool } from '../plugins/services/port-pool.js'
export type { ServiceConfig, ServiceStatus, ServiceStopSource } from '../plugins/services/schema.js'
Expand Down Expand Up @@ -77,6 +79,11 @@ export interface Platform {
* writing `sessions/<id>/calls/<callId>.json` and listing that directory.
*/
llmCallLog?: LLMCallStore
/**
* Shrinks images before they reach a model. Absent on hosts with a process table,
* which keep resizing with the `vipsheader` / `vipsthumbnail` binaries.
*/
imageResizer?: ImageResizer
/** Absolute path to the OS temp directory (equivalent to `os.tmpdir()` on Node/Bun). */
tmpDir: string
}
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ import { selectPluginState } from '~/core/sessions/reducer.js'
import { createTool } from '~/core/tools/definition.js'
import { ToolCallId } from '~/core/tools/schema.js'
import { toolEvents } from '~/core/tools/state.js'
import { Ok } from '~/lib/utils/result.js'
import { Err, Ok } from '~/lib/utils/result.js'
import { contextCompactPlugin } from '~/plugins/context-compact/index.js'
import { getAgentMailbox, selectMailboxState } from '~/plugins/mailbox/query.js'
import { mailboxEvents } from '~/plugins/mailbox/state.js'
Expand Down Expand Up @@ -320,6 +320,137 @@ describe('limits-guard plugin', () => {
await harness.shutdown()
})

it('a failing tool alternating with another successful tool → warned by name, then paused', async () => {
let inferenceCount = 0
let sawNamedWarning = false

const flakyTool = createTool({
name: 'flaky_tool',
description: 'Always fails.',
input: z.object({ attempt: z.number() }),
execute: async () => Err({ message: 'flaky_tool is broken', recoverable: true }),
})
const okTool = createTool({
name: 'ok_tool',
description: 'Always succeeds.',
input: z.object({ attempt: z.number() }),
execute: async () => Ok('fine'),
})

const harness = createLimitsHarness({
presets: [createMultiAgentPreset([
{
name: 'worker',
system: 'Worker agent.',
tools: [flakyTool, okTool],
agents: [],
plugins: [limitsGuardPlugin.configureAgent({
limits: { maxConsecutiveToolFailures: 3, maxRepeatedToolCalls: 10, maxTurns: 100 },
})],
},
], { orchestratorSystem: 'Orchestrator agent.' })],
mockHandler: (request) => {
if (request.systemPrompt.endsWith('Orchestrator agent.')) {
const hasStartedWorker = request.messages.some((message) =>
message.role === 'assistant'
&& message.toolCalls?.some((toolCall) => toolCall.name === 'start_worker')
)
return hasStartedWorker
? { content: 'Orchestration complete', toolCalls: [], finishReason: 'stop', metrics: MockLLMProvider.defaultMetrics() }
: {
content: null,
toolCalls: [{ id: ToolCallId('start-worker'), name: 'start_worker', input: { message: 'Do the work' } }],
finishReason: 'stop',
metrics: MockLLMProvider.defaultMetrics(),
}
}

inferenceCount++
if (JSON.stringify(request.messages).includes('Tool \\"flaky_tool\\" has failed 2 times in a row')) {
sawNamedWarning = true
}
const toolName = inferenceCount % 2 === 1 ? 'flaky_tool' : 'ok_tool'
return {
content: null,
toolCalls: [{ id: ToolCallId(`worker-${inferenceCount}`), name: toolName, input: { attempt: inferenceCount } }],
finishReason: 'stop',
metrics: MockLLMProvider.defaultMetrics(),
}
},
})

const session = await harness.createSession('test')
await session.sendMessage('Start')
await waitUntil(() => session.state.agents.get(AgentId('worker_1'))?.status === 'paused')

const worker = session.state.agents.get(AgentId('worker_1'))!
expect(worker.pauseMessage).toContain(`Tool 'flaky_tool' failed 3 consecutive times`)
expect(sawNamedWarning).toBe(true)
// flaky, ok, flaky, ok, flaky → the 6th inference sees the third failure and pauses.
expect(inferenceCount).toBe(6)

await harness.shutdown()
})

it('repeated identical tool call → agent paused before the failure limit', async () => {
let inferenceCount = 0

const failingTool = createTool({
name: 'failing_tool',
description: 'Always fails.',
input: z.object({ path: z.string() }),
execute: async () => Err({ message: 'failing_tool is broken', recoverable: true }),
})

const harness = createLimitsHarness({
presets: [createMultiAgentPreset([
{
name: 'worker',
system: 'Worker agent.',
tools: [failingTool],
agents: [],
plugins: [limitsGuardPlugin.configureAgent({
limits: { maxConsecutiveToolFailures: 5, maxRepeatedToolCalls: 4, maxTurns: 100 },
})],
},
], { orchestratorSystem: 'Orchestrator agent.' })],
mockHandler: (request) => {
if (request.systemPrompt.endsWith('Orchestrator agent.')) {
const hasStartedWorker = request.messages.some((message) =>
message.role === 'assistant'
&& message.toolCalls?.some((toolCall) => toolCall.name === 'start_worker')
)
return hasStartedWorker
? { content: 'Orchestration complete', toolCalls: [], finishReason: 'stop', metrics: MockLLMProvider.defaultMetrics() }
: {
content: null,
toolCalls: [{ id: ToolCallId('start-worker'), name: 'start_worker', input: { message: 'Do the work' } }],
finishReason: 'stop',
metrics: MockLLMProvider.defaultMetrics(),
}
}

inferenceCount++
return {
content: null,
toolCalls: [{ id: ToolCallId(`worker-${inferenceCount}`), name: 'failing_tool', input: { path: 'same.txt' } }],
finishReason: 'stop',
metrics: MockLLMProvider.defaultMetrics(),
}
},
})

const session = await harness.createSession('test')
await session.sendMessage('Start')
await waitUntil(() => session.state.agents.get(AgentId('worker_1'))?.status === 'paused')

const worker = session.state.agents.get(AgentId('worker_1'))!
expect(worker.pauseMessage).toBe('Repeated identical tool call detected (4 times)')
expect(inferenceCount).toBe(5)

await harness.shutdown()
})

it('send_message followed by non-communication work does not trip the no-progress limit', async () => {
let workerInferenceCount = 0
let workCount = 0
Expand Down
17 changes: 4 additions & 13 deletions packages/sdk/src/plugins/limits-guard/plugin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -122,18 +122,11 @@ export type BudgetExceededEvent = (typeof limitsEvents)['Events']['budget_exceed
// ============================================================================

/**
* Resolve tool name from agent state for tool_completed/tool_failed events.
* Must run BEFORE core reducer (which clears executingToolCall).
* Name of the tool behind a tool_completed/tool_failed event. Plugin reducers run after the
* core reducer, which has already cleared the running call and recorded the named result.
*/
function resolveToolName(
sessionState: { agents: Map<AgentId, { executingToolCall?: { toolName: string }; pendingToolCalls: Array<{ id: string; name: string }> }> },
agentId: AgentId,
toolCallId: string,
): string {
const agent = sessionState.agents.get(agentId)
if (!agent) return 'unknown'
return agent.executingToolCall?.toolName
?? agent.pendingToolCalls.find((tc) => tc.id === toolCallId)?.name
function resolveToolName(sessionState: SessionState, agentId: AgentId, toolCallId: ToolCallId): string {
return sessionState.agents.get(agentId)?.pendingToolResults.find((result) => result.toolCallId === toolCallId)?.toolName
?? 'unknown'
}

Expand Down Expand Up @@ -279,7 +272,6 @@ export const limitsGuardPlugin = definePlugin('limits-guard')
const counters = limits.get(event.agentId)
if (!counters) return limits

// Resolve tool name from agent state (must run before core clears executingToolCall)
const toolName = resolveToolName(sessionState, event.agentId, event.toolCallId)
const { [toolName]: _, ...restFailures } = counters.consecutiveToolFailures

Expand All @@ -296,7 +288,6 @@ export const limitsGuardPlugin = definePlugin('limits-guard')
const counters = limits.get(event.agentId)
if (!counters) return limits

// Resolve tool name from agent state (must run before core clears executingToolCall)
const toolName = resolveToolName(sessionState, event.agentId, event.toolCallId)
const currentEntry = counters.consecutiveToolFailures[toolName]

Expand Down
Loading
Loading