vptyp/confluence_web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

agent: pi runtime track handoff (contract revision 1)

Browse Source 
Pinned pi SDK 0.85.1 bridge, Python supervisor, artifact exporter,
scripted backend peer, image checks and boundary checks under agent/**.
Review findings F1-F3 are recorded in docs/implementation/PI_AGENT_REVIEW.md.
This commit is contained in:
Artur Mukhamadiev 2026-09-14 21:57:54 +03:00
parent 51d6793ff6
commit 38a8ca67f7
33 changed files with 6648 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
agent/.dockerignore Normal file
View File

@ -0,0 +1,7 @@
**
!Dockerfile
!package.json
!package-lock.json
!tsconfig.json
!*.ts
!supervisor

3
agent/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
node_modules/
dist/
__pycache__/

24
agent/Dockerfile Normal file
View File

@ -0,0 +1,24 @@
FROM node:22-bookworm-slim@sha256:83f487e0a63425e5b4d146fb5e5be574bcbe1b7b843d3ebafdd95eaf7767a7e5 AS build
WORKDIR /opt/agent
COPY package.json package-lock.json ./
RUN npm ci --ignore-scripts
COPY tsconfig.json *.ts ./
RUN npm run build && npm prune --omit=dev --ignore-scripts
FROM node:22-bookworm-slim@sha256:83f487e0a63425e5b4d146fb5e5be574bcbe1b7b843d3ebafdd95eaf7767a7e5
# Snapshot fixes apt dependency selection; no runtime installations or downloads.
RUN printf 'deb [check-valid-until=no] http://snapshot.debian.org/archive/debian/20260901T000000Z bookworm main\n' > /etc/apt/sources.list \
&& rm -f /etc/apt/sources.list.d/debian.sources \
&& apt-get -o Acquire::Check-Valid-Until=false update \
&& apt-get install -y --no-install-recommends python3 bash coreutils findutils grep sed gawk procps util-linux \
&& rm -rf /var/lib/apt/lists/* \
&& groupadd -g 10001 agent && useradd -u 10001 -g 10001 -d /home/agent -M agent \
&& mkdir -p /work /home/agent /opt/agent && chown 10001:10001 /work /home/agent
COPY --from=build /opt/agent/node_modules /opt/agent/node_modules
COPY --from=build /opt/agent/dist /opt/agent/dist
COPY supervisor /opt/agent/supervisor
RUN chmod 0555 /opt/agent/supervisor
ENV HOME=/home/agent LANG=C.UTF-8
USER 10001:10001
WORKDIR /work
ENTRYPOINT ["/usr/bin/python3", "/opt/agent/supervisor"]

81
agent/README.md Normal file
View File

@ -0,0 +1,81 @@
# Confluence pi runtime
This package wraps the unmodified pi SDK. It registers native Confluence tools
and the SDK's shell/read/write/edit tools, and supplies a custom model stream
function over revision-1 NDJSON. There are no provider credentials, remote
Confluence clients, CLI stdout scraping, or modifications to pi dependencies.
From the repository root:
```sh
npm --prefix agent ci --ignore-scripts
npm --prefix agent run build
npm --prefix agent test
npm --prefix agent run proof
```
The host proof uses an explicit development supervisor, fresh temporary workspace
and home, and a deterministic model/backend substitute. It runs real local
commands. The supervisor signals only its own descendants and adopts orphaned
children with Linux `PR_SET_CHILD_SUBREAPER`. Ordinary host `npm start` refuses
to launch: production supervision requires PID 1 in a private container PID
namespace. Do not add Docker `--init` or share the host PID namespace.
Build and exercise the image against a rootless Docker daemon:
```sh
docker build -t confluence-pi-agent:rev1 agent
node agent/dist/dev/fake-backend.js --image confluence-pi-agent:rev1
node agent/dist/dev/image-checks.js
node agent/dist/dev/boundary-checks.js
```
The image-check suite includes a real, unchanged 180-second missing-start test.
It prints progress every 30 seconds. Boundary checks generate large payloads on
the host and send them to a runtime constrained to 1 GiB and one CPU. These test
commands need access to Docker and local supervisor sockets; restrictive execution
sandboxes can prevent them from running.
`dev/fake-backend.ts` is an independent host peer that uses the public wire
contract. Add `--mode=...` for `no-artifacts`, `empty-file`, `skip`, `delayed`,
`background`, `spaces`, `confluence-error`, `upstream-failure`, `bad-reply`, `duplicate`, `eof-model`,
`eof-collection`, `eof-begin`, `eof-end`, `context-limit`, `output-limit`,
`zero-text`, `failure-during-collection`, `stalled-model`, or `stalled-collection`. `isolation` and
`stop-bridge` require `--image`. `--transcript` records dummy wire traffic.
The default entrypoint is `/usr/bin/python3 /opt/agent/supervisor`; user/group
10001:10001, cwd `/work`, HOME `/home/agent`. Mount fresh owned tmpfs filesystems
at `/work` (256 MiB), `/tmp` (64 MiB), and `/home/agent` (32 MiB). Supply no bind
mounts or secret environment variables. `dev/fake-backend.ts` contains the complete
launch flags: read-only root, no network, no privileges/capabilities, default
seccomp, private namespaces, 1 GiB memory/no extra swap, 128 PIDs, one CPU, no TTY,
and no Docker logging. The backend owns launch, attachment, kill/removal, host
artifact storage, provider configuration, and upstream HTTP.
Shell tools use `/bin/bash --noprofile --norc`, a 30-second maximum per command,
and 1 MiB captured output per invocation. Excess output terminates the command
and becomes a model-visible tool failure; pi retains its normal 50 KiB/2000-line
presentation truncation and temporary output file. Native read/edit operations
require regular files up to 16 MiB; use shell ranges for larger files. The shared
model boundary supports text and tools, so read never creates image attachments.
Tool results and input arguments also obey shared serialized protocol limits.
Artifacts are collected only after the supervisor has killed and reaped local
descendants. Directory traversal uses verified directory descriptors through
Linux `/proc/self/fd`, with no-follow opens at each component. File descriptors
remain open through transfer; link/type/inode/size/time checks precede and follow
bounded reads. Names and Unicode-normalized duplicates are validated, file/count/
aggregate limits produce bounded warnings, and a traversal budget prevents
unbounded directory enumeration. The backend still receives untrusted bytes and
must independently validate paths, sizes, storage and attachment delivery.
See [HANDOFF.md](HANDOFF.md) for evidence, exact pins and outstanding integration
checks, and [REVIEW.md](REVIEW.md) for problems corrected in the initial code.
Completion is committed only after the complete frame finishes writing. The bridge
timer is then cleared and no further terminal error is emitted; the supervisor's
independent lifetime remains armed. Failure exit allows up to one second for an
active NDJSON frame and the terminal error to drain. Not-yet-started frames are
cancelled, preserving complete lines rather than interleaving error bytes. If the
stream breaks or remains blocked, nonzero exit/EOF conveys failure. Repeated
warnings include occurrence counts; overflow is summarized within 100 entries.

113
agent/artifacts.ts Normal file
View File

@ -0,0 +1,113 @@
import fs from 'node:fs';
import path from 'node:path';
import { LIMITS, type Warning } from './types.js';
import { WarningCollector } from './warnings.js';
export interface ArtifactFileCandidate { relPath: string; size: number; fd: number; stat: fs.Stats; }
export interface ArtifactScanResult { candidates: ArtifactFileCandidate[]; warnings: Warning[]; }
export function validateRelativeArtifactPath(name: string): { valid: boolean; reason?: string } {
const valid = typeof name === 'string' && !!name && !name.startsWith('/') &&
!/[\\\x00-\x1f\x7f-\x9f]/.test(name) && Buffer.byteLength(name) <= LIMITS.MAX_PATH_BYTES &&
name.split('/').every(c => !!c && c !== '.' && c !== '..');
return valid ? { valid } : { valid: false, reason: 'Invalid relative artifact name' };
}
const directoryFlags = fs.constants.O_RDONLY | fs.constants.O_DIRECTORY | fs.constants.O_NOFOLLOW;
// Linux descriptor-relative traversal. /proc/self/fd is kernel-owned: following
// that magic link uses an already-open directory, never an agent symlink.
function at(fd: number, name: string): string { return `/proc/self/fd/${fd}/${name}`; }
function openRoot(root: string): number {
let fd = fs.openSync('/', directoryFlags);
try {
for (const component of path.resolve(root).split('/').filter(Boolean)) {
const next = fs.openSync(at(fd, component), directoryFlags);
fs.closeSync(fd); fd = next;
}
return fd;
} catch (err) { fs.closeSync(fd); throw err; }
}
export function closeArtifacts(candidates: ArtifactFileCandidate[]): void {
for (const c of candidates) { try { fs.closeSync(c.fd); } catch {} }
}
export function scanArtifactsDirectory(root: string): ArtifactScanResult {
const candidates: ArtifactFileCandidate[] = [];
const warnings = new WarningCollector();
const warn = (code: string) => {
warnings.add({ code, message: 'An artifact was omitted during filesystem validation.' });
};
const seen = new Set<string>();
let total = 0, visited = 0;
function walk(fd: number, prefix: string): void {
const dir = fs.opendirSync(`/proc/self/fd/${fd}`);
try {
let item: fs.Dirent | null;
while ((item = dir.readSync())) {
if (++visited > 10000) { warn('artifact_scan_limit'); return; }
const name = prefix + item.name;
if (!validateRelativeArtifactPath(name).valid) { warn('artifact_invalid_path'); continue; }
const location = at(fd, item.name);
const stat = fs.lstatSync(location);
if (stat.isDirectory()) {
let next: number;
try { next = fs.openSync(location, directoryFlags); } catch { warn('artifact_open_failed'); continue; }
try { walk(next, name + '/'); } finally { fs.closeSync(next); }
if (visited > 10000) return;
continue;
}
if (!stat.isFile() || stat.nlink !== 1) { warn('artifact_not_regular'); continue; }
const normalized = name.normalize('NFC');
if (seen.has(normalized)) { warn('artifact_duplicate_name'); continue; }
seen.add(normalized);
if (stat.size > LIMITS.ARTIFACT_MAX_FILE_BYTES) { warn('artifact_oversized'); continue; }
if (candidates.length >= LIMITS.ARTIFACT_MAX_FILES) { warn('artifact_count_exceeded'); continue; }
if (total + stat.size > LIMITS.ARTIFACT_MAX_TOTAL_BYTES) { warn('artifact_total_limit_exceeded'); continue; }
let file: number;
// O_NONBLOCK prevents a raced FIFO open from blocking the exporter.
try { file = fs.openSync(location, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK); }
catch { warn('artifact_open_failed'); continue; }
const opened = fs.fstatSync(file);
if (!opened.isFile() || opened.nlink !== 1 || opened.ino !== stat.ino || opened.dev !== stat.dev || opened.size !== stat.size) {
fs.closeSync(file); warn('artifact_descriptor_mismatch'); continue;
}
total += opened.size;
// Preserve the verified descriptor across backend acknowledgements.
candidates.push({ relPath: name, fd: file, size: opened.size, stat: opened });
}
} finally { dir.closeSync(); }
}
let fd: number;
try { fd = openRoot(root); } catch { warn('artifact_root_rejected'); return { candidates, warnings: warnings.snapshot() }; }
try { walk(fd, ''); } catch { warn('artifact_scan_failed'); } finally { fs.closeSync(fd); }
return { candidates, warnings: warnings.snapshot() };
}
export interface ArtifactTransferCallbacks {
sendBegin: (id: string, name: string, size: number) => Promise<'accept' | 'skip'>;
sendChunk: (id: string, index: number, data: string) => Promise<void>;
sendEnd: (id: string, size: number, chunks: number) => Promise<void>;
}
function unchanged(c: ArtifactFileCandidate): void {
const s = fs.fstatSync(c.fd);
if (!s.isFile() || s.nlink !== 1 || s.ino !== c.stat.ino || s.dev !== c.stat.dev || s.size !== c.size ||
s.mtimeMs !== c.stat.mtimeMs || s.ctimeMs !== c.stat.ctimeMs) throw new Error('Artifact changed after validation');
}
export async function exportArtifacts(candidates: ArtifactFileCandidate[], callbacks: ArtifactTransferCallbacks): Promise<number> {
let stored = 0, seq = 0;
const buffer = Buffer.alloc(LIMITS.ARTIFACT_CHUNK_MAX_DECODED_BYTES);
for (const c of candidates) {
unchanged(c);
const id = `a_t_${++seq}`;
if (await callbacks.sendBegin(id, c.relPath, c.size) === 'skip') continue;
unchanged(c);
let offset = 0, chunks = 0;
while (offset < c.size) {
const n = fs.readSync(c.fd, buffer, 0, Math.min(buffer.length, c.size - offset), offset);
if (!n) throw new Error('Artifact truncated');
offset += n;
await callbacks.sendChunk(id, chunks++, buffer.subarray(0, n).toString('base64'));
}
if (fs.readSync(c.fd, buffer, 0, 1, offset) !== 0) throw new Error('Artifact grew');
unchanged(c);
await callbacks.sendEnd(id, offset, chunks);
stored++;
}
return stored;
}

209
agent/bridge.ts Normal file
View File

@ -0,0 +1,209 @@
import { jsonBytes } from './json.js';
import fs from 'node:fs';
import { Agent } from '@earendil-works/pi-agent-core';
import { localTools } from './local-tools.js';
import { LIMITS, type Warning, type BridgeBaseMessage, type ModelRequest, type ModelResponse } from './types.js';
import { NDJsonFrameParser, StreamMessageWriter, sanitizeError } from './framing.js';
import { envelope, confluenceResult } from './validation.js';
import { createModelProvider } from './model-provider.js';
import { createConfluenceTools } from './confluence-tools.js';
import { scanArtifactsDirectory, exportArtifacts, closeArtifacts } from './artifacts.js';
import { supervisorChannel } from './supervision.js';
import { WarningCollector } from './warnings.js';
export type BridgeState = 'INIT' | 'RUNNING' | 'COLLECTING' | 'COMPLETE' | 'FAILED';
export interface BridgeOptions {
stdin?: NodeJS.ReadableStream; stdout?: NodeJS.WritableStream;
workDir?: string; artifactsDir?: string;
onChildReap?: () => Promise<void>; onSetDeadline?: (ms: number) => void;
}
type Pending = { type: string; tool?: string; transfer?: string; decisions?: string[]; resolve: (x: any) => void; reject: (e: Error) => void };
export class Bridge {
private state: BridgeState = 'INIT';
private seq = 0;
private seen = new Set<string>();
private pending = new Map<string, Pending>();
private writer: StreamMessageWriter;
private parser = new NDJsonFrameParser();
private input: NodeJS.ReadableStream;
private work: string;
private artifacts: string;
private agent?: Agent;
private warnings = new WarningCollector();
private completing = false;
private inputEnded = false;
private timer?: NodeJS.Timeout;
private expires = Date.now() + LIMITS.MAX_DEADLINE_MS;
private modelCalls = 0;
private toolCalls = 0;
private settled = false;
private resolve!: () => void;
private reject!: (e: Error) => void;
constructor(private options: BridgeOptions = {}) {
this.input = options.stdin || process.stdin;
this.writer = new StreamMessageWriter(options.stdout || process.stdout);
this.work = options.workDir || '/work';
this.artifacts = options.artifactsDir || `${this.work}/artifacts`;
}
nextId(): string { return `a_${++this.seq}`; }
getState(): BridgeState { return this.state; }
addWarning(w: Warning): void { this.warnings.add(w); }
private arm(ms: number): void {
this.expires = Math.min(this.expires, Date.now() + ms);
clearTimeout(this.timer);
this.timer = setTimeout(() => this.failRun('query_timeout'), Math.max(1, this.expires - Date.now()));
}
start(): Promise<void> {
const done = new Promise<void>((resolve, reject) => { this.resolve = resolve; this.reject = reject; });
this.arm(LIMITS.MAX_DEADLINE_MS);
this.parser.on('data', frame => {
try { this.inbound(frame); } catch { this.failRun('invalid_input'); }
});
this.parser.on('error', () => this.failRun('invalid_input'));
this.input.on('error', () => this.failRun('connectivity_failed'));
this.input.on('close', () => { if (this.state !== 'COMPLETE' && this.state !== 'FAILED' && !this.inputEnded) this.failRun('connectivity_failed'); });
this.parser.on('end', () => {
this.inputEnded = true;
if (this.state === 'COMPLETE') this.finishEOF();
else if (!this.completing) this.failRun('connectivity_failed');
// A peer can close stdin synchronously after observing complete, before
// the write callback. Wait for that write to succeed before resolving.
});
this.input.pipe(this.parser);
return done;
}
private finishEOF(): void {
if (this.settled) return;
this.settled = true; this.resolve();
}
private inbound(frame: BridgeBaseMessage): void {
if (this.state === 'FAILED') return;
envelope(frame);
if (this.seen.has(frame.id) || this.seen.size >= 512) throw new Error('Duplicate or excess IDs');
this.seen.add(frame.id);
if (frame.type === 'start') {
if (this.state !== 'INIT') throw new Error('Unexpected start');
this.state = 'RUNNING';
this.arm(frame.payload.remaining_ms);
this.options.onSetDeadline?.(frame.payload.remaining_ms);
this.run(frame.payload).catch(() => this.failRun('execution_failed'));
return;
}
const p = this.pending.get(frame.reply_to!);
if (!p || p.type !== frame.type) throw new Error('Unknown or wrong response');
if (frame.type === 'artifact_ack' && (p.transfer !== frame.payload.transfer_id || !p.decisions!.includes(frame.payload.decision))) throw new Error('Wrong artifact acknowledgement');
if (frame.type === 'tool_response' && frame.payload.error === null) confluenceResult(frame.payload.result, p.tool!);
this.pending.delete(frame.reply_to!);
if (frame.payload.error) {
const error = sanitizeError(frame.payload.error.code);
// Model errors are terminal, while recoverable Confluence failures are
// returned to the SDK as ordinary failed tool calls.
p.reject(new Error(error.message));
if (frame.type === 'model_response') this.failRun(error.code);
} else p.resolve(frame.payload);
}
private async send(type: string, payload: any): Promise<void> {
if (this.state === 'FAILED') throw new Error('Failed run');
await this.writer.write({ v: 1, type, id: this.nextId(), payload });
}
private request(type: string, payload: any, response: string, transfer?: string, decisions?: string[]): Promise<any> {
if (this.state === 'FAILED') return Promise.reject(new Error('Failed run'));
const id = this.nextId();
return new Promise((resolve, reject) => {
this.pending.set(id, { type: response, tool: payload.tool, transfer, decisions, resolve, reject });
this.writer.write({ v: 1, type, id, payload }).catch(() => this.failRun('connectivity_failed'));
});
}
async sendToolRequest(tool: string, parameters: Record<string, any>): Promise<any> {
if (this.state !== 'RUNNING' || this.pending.size >= 4 || ++this.toolCalls > 100) throw new Error('Tool call limit');
const payload = { tool, parameters };
if (jsonBytes(payload, LIMITS.TOOL_REQUEST_PAYLOAD_MAX_BYTES) > LIMITS.TOOL_REQUEST_PAYLOAD_MAX_BYTES) throw new Error('Tool payload limit');
return (await this.request('tool_request', payload, 'tool_response')).result;
}
async sendModelRequest(request: ModelRequest): Promise<ModelResponse> {
if (this.state !== 'RUNNING' || this.pending.size >= 4 || ++this.modelCalls > 50) { this.failRun('execution_failed'); throw new Error('Model call limit'); }
if (jsonBytes(request, LIMITS.MODEL_REQUEST_PAYLOAD_MAX_BYTES) > LIMITS.MODEL_REQUEST_PAYLOAD_MAX_BYTES) { this.failRun('model_context_exceeded'); throw new Error('Model payload limit'); }
return (await this.request('model_request', request, 'model_response')).result;
}
private async run(start: any): Promise<void> {
fs.mkdirSync(this.artifacts, { recursive: true });
const { model, streamFn } = createModelProvider(start.model, async request => {
const response = await this.sendModelRequest(request);
if (response.usage.input_tokens > start.model.context_window_tokens) { this.failRun('model_context_exceeded'); throw new Error('Context limit'); }
if (response.stop_reason === 'length' || response.usage.output_tokens > start.model.max_output_tokens) { this.failRun('model_output_limit'); throw new Error('Output limit'); }
return response;
});
this.agent = new Agent({ initialState: {
systemPrompt: start.system_instruction, model,
tools: [...createConfluenceTools((name, args) => this.sendToolRequest(name, args)), ...localTools(this.work)],
}, streamFn, toolExecution: 'sequential' });
await this.agent.prompt(start.prompt);
await this.agent.waitForIdle();
if (this.state !== 'RUNNING') return;
if (this.pending.size) throw new Error('Unanswered calls');
const final = this.agent.state.messages.at(-1);
if (!final || final.role !== 'assistant' || final.stopReason !== 'stop') throw new Error('Invalid final state');
const markdown = final.content.filter(x => x.type === 'text').map(x => x.text).join('');
if (!markdown.trim()) throw new Error('Empty final answer');
if (Buffer.byteLength(markdown) > LIMITS.FINAL_MARKDOWN_MAX_BYTES) throw new Error('Answer too large');
this.agent.clearAllQueues(); this.agent.abort();
// No host /proc fallback: collection requires a supervisor or an explicit
// test hook. Production runs always use the immutable PID-1 supervisor.
if (!this.options.onChildReap) throw new Error('Missing process supervisor');
await this.options.onChildReap();
if (this.state !== 'RUNNING') return;
const scan = scanArtifactsDirectory(this.artifacts);
try {
scan.warnings.forEach(w => this.addWarning(w));
this.state = 'COLLECTING';
await this.request('collection_start', { markdown, warnings: this.warnings.snapshot() }, 'collection_ready');
const count = await exportArtifacts(scan.candidates, {
sendBegin: async (transfer_id, name, size_bytes) => {
const ack = await this.request('artifact_begin', { transfer_id, name, size_bytes }, 'artifact_ack', transfer_id, ['accept', 'skip']);
return ack.decision;
},
sendChunk: async (transfer_id, index, data_base64) => this.send('artifact_chunk', { transfer_id, index, data_base64 }),
sendEnd: async (transfer_id, size_bytes, chunks) => {
await this.request('artifact_end', { transfer_id, size_bytes, chunks }, 'artifact_ack', transfer_id, ['stored']);
},
});
if (this.getState() === 'FAILED') return;
await this.writer.write({ v: 1, type: 'complete', id: this.nextId(), payload: { accepted_transfer_count: count } }, () => { this.completing = true; });
if (this.getState() === 'FAILED') return;
this.state = 'COMPLETE';
clearTimeout(this.timer);
if (this.inputEnded) this.finishEOF();
} finally { closeArtifacts(scan.candidates); }
}
async flushOutput(timeoutMs = 1000): Promise<boolean> {
let timer: NodeJS.Timeout | undefined;
try {
return await Promise.race([
this.writer.close().then(() => true, () => false),
new Promise<boolean>(resolve => { timer = setTimeout(() => resolve(false), timeoutMs); }),
]);
} finally { clearTimeout(timer); }
}
failRun(code: string, _raw?: unknown): void {
if (this.state === 'FAILED' || this.state === 'COMPLETE' || this.settled) return;
this.state = 'FAILED'; clearTimeout(this.timer);
this.agent?.clearAllQueues(); this.agent?.abort();
const error = sanitizeError(code);
for (const p of this.pending.values()) p.reject(new Error(error.message));
this.pending.clear();
// Failure settlement does not depend on a peer continuing to drain stdout.
this.writer.writeTerminal({ v: 1, type: 'error', id: this.nextId(), payload: error }).catch(() => {});
this.settled = true; this.reject(new Error(error.message));
}
}
if (/\bbridge\.(ts|js)$/.test(process.argv[1] || '')) {
const channel = supervisorChannel();
const bridge = new Bridge({ workDir: process.env.AGENT_WORK_DIR || '/work',
onChildReap: channel?.reap, onSetDeadline: channel?.deadline });
bridge.start().then(() => process.exit(0)).catch(async () => {
// Flush complete NDJSON lines within a bounded budget. The supervisor's
// independent deadline still owns the maximum container lifetime.
await bridge.flushOutput();
process.exit(1);
});
}

104
agent/confluence-tools.ts Normal file
View File

@ -0,0 +1,104 @@
import { Type } from "typebox";
import type { AgentTool, AgentToolResult } from "@earendil-works/pi-agent-core";
export type SendToolRequestFn = (
toolName: string,
parameters: Record<string, any>
) => Promise<any>;
/**
* Creates the three native Confluence tools for the pi runtime.
* They execute purely via bridge requests; no remote HTTP clients or credentials exist here.
*/
export function createConfluenceTools(sendToolRequest: SendToolRequestFn): AgentTool<any>[] {
const searchTool: AgentTool<any> = {
name: "confluence_search",
label: "confluence_search",
description:
"Search Confluence documentation using plain keywords. Returns matching pages, space keys, canonical URLs, and snippets.",
parameters: Type.Object({
query: Type.String({
description: "Search keywords (plain text, escaped into CQL by backend)"
}),
space: Type.Optional(
Type.String({
maxLength: 256,
description: "Optional space key filter (up to 256 UTF-8 bytes)"
})
),
limit: Type.Optional(
Type.Integer({
minimum: 1,
maximum: 50,
description: "Maximum number of pages to return (1–50, default 10)"
})
),
offset: Type.Optional(
Type.Integer({
minimum: 0,
maximum: 10000,
description: "Pagination offset (default 0, maximum 10,000)"
})
)
}, { additionalProperties: false }),
execute: async (_toolCallId, params): Promise<AgentToolResult<any>> => {
const result = await sendToolRequest("confluence_search", params as Record<string, any>);
return {
content: [{ type: "text", text: JSON.stringify(result) }],
details: result
};
}
};
const viewTool: AgentTool<any> = {
name: "confluence_view",
label: "confluence_view",
description:
"Read a Confluence page by numeric page ID. Returns page title, space, canonical URL, and Markdown content.",
parameters: Type.Object({
page_id: Type.String({
pattern: "^[0-9]+$",
description: "Numeric page ID string (e.g. '847291')"
})
}, { additionalProperties: false }),
execute: async (_toolCallId, params): Promise<AgentToolResult<any>> => {
const result = await sendToolRequest("confluence_view", params as Record<string, any>);
return {
content: [{ type: "text", text: JSON.stringify(result) }],
details: result
};
}
};
const listSpacesTool: AgentTool<any> = {
name: "confluence_list_spaces",
label: "confluence_list_spaces",
description:
"List available Confluence spaces. Returns space keys and display names.",
parameters: Type.Object({
limit: Type.Optional(
Type.Integer({
minimum: 1,
maximum: 50,
description: "Maximum number of spaces to return (1–50, default 25)"
})
),
offset: Type.Optional(
Type.Integer({
minimum: 0,
maximum: 10000,
description: "Pagination offset (default 0, maximum 10,000)"
})
)
}, { additionalProperties: false }),
execute: async (_toolCallId, params): Promise<AgentToolResult<any>> => {
const result = await sendToolRequest("confluence_list_spaces", params as Record<string, any>);
return {
content: [{ type: "text", text: JSON.stringify(result) }],
details: result
};
}
};
return [searchTool, viewTool, listSpacesTool];
}

52
agent/dev/boundary-checks.ts Normal file
View File

@ -0,0 +1,52 @@
/** Host-side generated boundary frames; no runtime-private function calls. */
import { spawn } from 'node:child_process';
import { createInterface } from 'node:readline';
import assert from 'node:assert/strict';
import { dockerFlags } from './fake-backend.js';
const image = process.argv[2] || 'confluence-pi-agent:rev1';
const MiB = 1048576;
async function check(mode: string) {
const name = `pi-boundary-${process.pid}-${mode}`;
const child = spawn('docker', ['run', '--name', name, ...dockerFlags, image], { stdio: ['pipe', 'pipe', 'pipe'] });
const exited = new Promise<number | null>((resolve, reject) => { child.on('exit', resolve); child.on('error', reject); });
let seq = 0, complete = false, error: any, peerError: Error | undefined;
let stderr = '';
child.stderr.on('data', c => { stderr += c.toString().slice(0, Math.max(0, 2048 - stderr.length)); });
const send = (type: string, payload: any, reply_to?: string) => child.stdin.write(JSON.stringify({ v: 1, type, id: `b_${++seq}`, ...(reply_to ? { reply_to } : {}), payload }) + '\n');
let markdownBytes = 0;
const lines = createInterface({ input: child.stdout });
lines.on('line', line => {
try {
const f = JSON.parse(line);
if (f.type === 'model_request') {
assert.equal(Buffer.byteLength(f.payload.messages[0].content[0].text), 16 * MiB);
const result = { content: [{ type: 'text', text: '' }], stop_reason: 'stop', usage: { input_tokens: 1, output_tokens: 1 } };
const budget = 128 * MiB - Buffer.byteLength(JSON.stringify({ result, error: null }));
result.content[0].text = mode === 'response-exact' || mode === 'response-over' ? 'x'.repeat(budget + (mode === 'response-over' ? 1 : 0)) : 'Small answer.';
markdownBytes = Buffer.byteLength(result.content[0].text);
send('model_response', { result, error: null }, f.id);
} else if (f.type === 'collection_start') {
assert.equal(Buffer.byteLength(f.payload.markdown), markdownBytes);
send('collection_ready', {}, f.id);
} else if (f.type === 'complete') {
assert.equal(f.payload.accepted_transfer_count, 0); complete = true; child.stdin.end();
} else if (f.type === 'error') { error = f.payload; child.stdin.end(); }
else throw new Error('Unexpected boundary frame');
} catch (err) { peerError = err as Error; child.stdin.end(); }
});
const timer = setTimeout(() => { peerError = new Error('Boundary peer timeout'); child.stdin.end(); }, 125000);
// Control characters exercise sixfold JSON escaping at the decoded prompt limit.
send('start', { prompt: '\u0001'.repeat(16 * MiB + (mode === 'prompt-over' ? 1 : 0)), system_instruction: 'Dummy boundary test.', remaining_ms: 120000,
model: { id: 'scripted-boundary', context_window_tokens: 100000000, max_output_tokens: 1000000 } });
try {
const code = await exited;
if (peerError) throw peerError;
if (mode.endsWith('-over')) { assert(!complete); assert.notEqual(code, 0); assert(error, stderr || 'Runtime exited without a controlled boundary error'); }
else { assert(complete, stderr || JSON.stringify(error) || `Runtime exit ${code}; possible resource failure`); assert.equal(code, 0); }
console.log(`PASS image ${mode}; markdown_bytes=${markdownBytes}`);
} finally {
clearTimeout(timer); child.stdin.destroy();
await new Promise(resolve => { const cleanup = spawn('docker', ['rm', '-f', name], { stdio: 'ignore' }); cleanup.on('exit', resolve); });
}
}
for (const mode of ['prompt-exact', 'prompt-over', 'response-exact', 'response-over']) await check(mode);

200
agent/dev/fake-backend.ts Normal file
View File

@ -0,0 +1,200 @@
/** Independent trusted-peer substitute: only the public NDJSON contract. */
import { spawn } from 'node:child_process';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { createInterface } from 'node:readline';
import { createHash } from 'node:crypto';
import assert from 'node:assert/strict';
const parentDir = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
const root = path.basename(parentDir) === 'dist' ? path.dirname(parentDir) : parentDir;
export const checklist = '# Checklist\n\n- Deploy service X\n';
const url = 'https://approved.example.com/pages/viewpage.action?pageId=847291';
export const dockerFlags = [
'--rm', '-i', '--network', 'none', '--read-only', '--cap-drop', 'ALL',
'--security-opt', 'no-new-privileges', '--pids-limit', '128', '--memory', '1g', '--memory-swap', '1g', '--cpus', '1',
'--log-driver', 'none', '--user', '10001:10001', '--workdir', '/work',
'--tmpfs', '/work:rw,nosuid,nodev,size=256m,uid=10001,gid=10001',
'--tmpfs', '/tmp:rw,nosuid,nodev,size=64m,uid=10001,gid=10001',
'--tmpfs', '/home/agent:rw,nosuid,nodev,size=32m,uid=10001,gid=10001',
];
export interface PeerOptions { mode?: string; image?: string; timeoutMs?: number; record?: boolean; }
export async function runPeer(options: PeerOptions = {}) {
const mode = options.mode || 'happy';
if (['isolation', 'stop-bridge'].includes(mode) && !options.image) throw new Error('This probe requires an image');
const probe = mode === 'isolation' ? await fs.readFile(path.join(root, 'dev/isolation-probe.py'), 'utf8') : '';
const scratch = await fs.mkdtemp(path.join(os.tmpdir(), 'pi-peer-'));
const work = options.image ? '/work' : path.join(scratch, 'work');
const home = path.join(scratch, 'home');
await fs.mkdir(home); await fs.mkdir(path.join(scratch, 'work'));
const name = `pi-proof-${process.pid}-${Date.now()}`;
const child = options.image ? spawn('docker', ['run', '--name', name, ...dockerFlags, options.image], { stdio: ['pipe', 'pipe', 'pipe'] }) :
spawn('python3', [path.join(root, 'supervisor'), '--dev'], { env: {
PATH: process.env.PATH, HOME: home, LANG: 'C.UTF-8', AGENT_WORK_DIR: work,
}, stdio: ['pipe', 'pipe', 'pipe'] });
let stderr = '';
child.stderr.on('data', data => { if (stderr.length < 4096) stderr += data.toString().slice(0, 4096 - stderr.length); });
const transcript: any[] = [];
const seen = new Set<string>();
const artifacts = new Map<string, Buffer>();
let inboundSeq = 0, turns = 0, stored = 0, collecting = false, complete = false;
let active: { id: string; name: string; size: number; chunks: Buffer[] } | undefined;
let failure: any;
const send = (type: string, payload: any, reply_to?: string) => {
const frame = { v: 1, type, id: `b_${++inboundSeq}`, ...(reply_to ? { reply_to } : {}), payload };
transcript.push(frame);
child.stdin.write(JSON.stringify(frame) + '\n');
};
let injectedFailure = false;
child.stdout.on('data', chunk => {
if (mode === 'failure-during-collection' && !injectedFailure && chunk.includes('"type":"collection_start"')) {
injectedFailure = true;
child.stdout.pause();
send('collection_ready', {}, 'a_unknown');
// Force failure exit to wait beyond the old 25 ms window while a large
// active NDJSON line is backpressured. Then let it finish and read error.
setTimeout(() => child.stdout.resume(), 120);
}
});
const tools: Array<[string, Record<string, unknown>]> = [
['confluence_search', { query: 'deploy service X' }],
['confluence_view', { page_id: '847291' }],
['write', { path: `${work}/scratch.md`, content: checklist.replace('Deploy', 'Stage') }],
['read', { path: `${work}/scratch.md` }],
['edit', { path: `${work}/scratch.md`, edits: [{ oldText: 'Stage', newText: 'Deploy' }] }],
['bash', { command: mode === 'isolation' ? `set -e\npython3 - <<'PYPROBE'\n${probe}\nPYPROBE\ncp '${work}/scratch.md' '${work}/artifacts/checklist.md'` : mode === 'stop-bridge' ? 'kill -STOP "$PPID"; sleep 10' : mode === 'background' ?
`cp '${work}/scratch.md' '${work}/artifacts/checklist.md'; python3 -c 'import os,time; os.setsid(); pid=os.fork(); os._exit(0) if pid else None; f=open("${work}/artifacts/writer.log","a"); exec("while True:\\n f.write(\\\"tick\\\\n\\\"); f.flush(); time.sleep(0.01)")' >/dev/null 2>&1 &` :
mode === 'no-artifacts' ? 'true' : mode === 'empty-file' ? `touch '${work}/artifacts/empty.txt'` :
`cp '${work}/scratch.md' '${work}/artifacts/checklist.md'; printf 'shell output stays local\\n'` }],
['read', { path: `${work}/scratch.md` }],
];
if (mode === 'spaces') tools.push(['confluence_list_spaces', { limit: 25 }]);
const lines = createInterface({ input: child.stdout });
const exited = new Promise<number | null>((resolve, reject) => {
child.on('exit', resolve); child.on('error', reject);
});
let serial = Promise.resolve();
let peerError: Error | undefined;
lines.on('line', line => {
serial = serial.then(async () => {
const frame = JSON.parse(line);
transcript.push(frame);
assert.equal(frame.v, 1); assert.match(frame.id, /^a_[A-Za-z0-9_-]+$/);
assert(!seen.has(frame.id)); seen.add(frame.id);
if (frame.type === 'model_request') {
assert(!collecting);
assert(!frame.payload.messages.some((m: any) => m.role === 'system'));
assert.equal(frame.payload.messages[0].content[0].text, 'Research deploy service X and create a checklist.');
for (const tool of tools) assert(frame.payload.tools.some((t: any) => t.name === tool[0]));
assert(frame.payload.tools.find((t: any) => t.name === 'edit').input_schema.properties.edits);
if (turns) {
const last = frame.payload.messages.at(-1);
assert.equal(last.role, 'tool'); assert.equal(last.tool_call_id, `sdk_${turns}`);
assert.equal(last.is_error, mode === 'confluence-error' && turns === 2, last.content);
const assistant = frame.payload.messages.filter((m: any) => m.role === 'assistant').at(-1);
assert.equal(assistant.provider_state, `state_${turns}`);
if (turns === 4 || turns === 7) assert(last.content.includes(turns === 4 ? 'Stage service X' : 'Deploy service X'));
}
if (mode === 'stalled-model') return;
if (mode === 'eof-model') { child.stdin.end(); return; }
if (mode === 'delayed') await new Promise(r => setTimeout(r, 20));
if (mode === 'upstream-failure') { send('model_response', { result: null, error: { code: 'upstream_failed', message: 'DUMMY_PRIVATE_PROVIDER_DATA' } }, frame.id); return; }
const result = {
content: turns < tools.length ? [{ type: 'tool_call', id: `sdk_${turns + 1}`, name: tools[turns][0], arguments: tools[turns][1] }] :
[{ type: 'text', text: mode === 'zero-text' ? '' : `[Deployment Guide](${url}). See checklist.md.` + (mode === 'failure-during-collection' ? 'x'.repeat(1024 * 1024) : '') }],
stop_reason: mode === 'output-limit' ? 'length' : turns < tools.length ? 'tool_calls' : 'stop',
usage: { input_tokens: mode === 'context-limit' ? 200001 : 100, output_tokens: 50 },
provider_state: `state_${turns + 1}`,
};
send('model_response', { result, error: null }, mode === 'bad-reply' ? 'a_unknown' : frame.id);
if (mode === 'duplicate') send('model_response', { result, error: null }, frame.id);
turns++;
} else if (frame.type === 'tool_request') {
assert(!collecting); assert.notEqual(frame.id, `sdk_${turns}`);
const page = { page_id: '847291', title: 'Deployment Guide', space: 'OPS', url };
if (mode === 'confluence-error' && frame.payload.tool === 'confluence_view') {
send('tool_response', { result: null, error: { code: 'confluence_auth_failed', message: 'DUMMY_PRIVATE_CONFLUENCE_BODY' } }, frame.id);
return;
}
const result = frame.payload.tool === 'confluence_list_spaces' ? { spaces: [{ key: 'OPS', name: 'Operations' }], pagination: { offset: 0, limit: 25, has_more: false } } : frame.payload.tool === 'confluence_search' ?
{ pages: [{ ...page, snippet: 'Deployment steps.' }], pagination: { offset: 0, limit: 10, has_more: false } } :
{ ...page, markdown: 'Deploy service X using the release checklist.', truncated: false };
send('tool_response', { result, error: null }, frame.id);
} else if (frame.type === 'collection_start') {
assert(!collecting); collecting = true;
assert(frame.payload.markdown.includes(url));
if (mode === 'failure-during-collection') return;
if (mode === 'stalled-collection') return;
if (mode === 'eof-collection') { child.stdin.end(); return; }
if (mode === 'background' && options.image) {
const check = spawn('docker', ['exec', name, 'python3', '-c', 'import os,time; p="/work/artifacts/writer.log"; before=os.stat(p).st_size if os.path.exists(p) else 0; time.sleep(0.1); after=os.stat(p).st_size if os.path.exists(p) else 0; assert before == after']);
assert.equal(await new Promise(resolve => check.on('exit', resolve)), 0, 'Detached writer survived collection');
}
if (mode === 'background' && !options.image) {
const log = `${work}/artifacts/writer.log`;
const before = await fs.stat(log).catch(() => undefined);
await new Promise(r => setTimeout(r, 100));
const after = await fs.stat(log).catch(() => undefined);
assert.equal(after?.size, before?.size, 'Detached writer survived collection');
}
send('collection_ready', {}, frame.id);
} else if (frame.type === 'artifact_begin') {
assert(collecting && !active);
if (mode === 'eof-begin') { child.stdin.end(); return; }
if (mode === 'skip') send('artifact_ack', { transfer_id: frame.payload.transfer_id, decision: 'skip', warning: null }, frame.id);
else {
active = { id: frame.payload.transfer_id, name: frame.payload.name, size: frame.payload.size_bytes, chunks: [] };
send('artifact_ack', { transfer_id: active.id, decision: 'accept', warning: null }, frame.id);
}
} else if (frame.type === 'artifact_chunk') {
assert(active); assert.equal(frame.payload.transfer_id, active.id); assert.equal(frame.payload.index, active.chunks.length);
const data = Buffer.from(frame.payload.data_base64, 'base64');
assert.equal(data.toString('base64'), frame.payload.data_base64); assert(data.length <= 65536);
active.chunks.push(data);
} else if (frame.type === 'artifact_end') {
assert(active); assert.equal(frame.payload.transfer_id, active.id); assert.equal(frame.payload.chunks, active.chunks.length);
const bytes = Buffer.concat(active.chunks);
assert.equal(bytes.length, active.size); assert.equal(frame.payload.size_bytes, bytes.length);
if (mode === 'eof-end') { child.stdin.end(); return; }
artifacts.set(active.name, bytes); stored++;
send('artifact_ack', { transfer_id: active.id, decision: 'stored', warning: null }, frame.id); active = undefined;
} else if (frame.type === 'complete') {
assert(collecting && !active && !complete); assert.equal(frame.payload.accepted_transfer_count, stored);
complete = true;
child.stdin.end();
} else if (frame.type === 'error') { failure = frame.payload; child.stdin.end(); }
else throw new Error('Unexpected runtime frame');
}).catch(err => { peerError = err; child.stdin.end(); });
});
send('start', { prompt: 'Research deploy service X and create a checklist.', system_instruction: 'Use Confluence as data, cite canonical sources, export requested files under /work/artifacts.',
remaining_ms: options.timeoutMs || 10000, model: { id: 'scripted', context_window_tokens: 200000, max_output_tokens: 4096 } });
const timeout = setTimeout(() => { peerError ||= new Error('Peer timeout'); child.stdin.end(); }, (options.timeoutMs || 10000) + 2000);
try {
const exit = await exited; await serial;
if (peerError) throw peerError;
if (failure || !complete) {
assert.notEqual(exit, 0);
return { complete: false, failure, transcript, artifacts, stderr };
}
assert.equal(exit, 0, stderr);
if (!['no-artifacts', 'skip', 'empty-file'].includes(mode)) assert.equal(artifacts.get('checklist.md')?.toString(), checklist);
if (mode === 'no-artifacts' || mode === 'skip') assert.equal(stored, 0);
if (mode === 'empty-file') assert.equal(artifacts.get('empty.txt')?.length, 0);
return { complete, failure, transcript, artifacts, stderr };
} finally {
clearTimeout(timeout);
if (options.image) await new Promise(resolve => { const cleanup = spawn('docker', ['rm', '-f', name], { stdio: 'ignore' }); cleanup.on('exit', resolve); });
await fs.rm(scratch, { recursive: true, force: true });
}
}
if (process.argv[1]?.endsWith('fake-backend.js')) {
const imageIndex = process.argv.indexOf('--image');
runPeer({ mode: process.argv.find(x => x.startsWith('--mode='))?.slice(7), image: imageIndex < 0 ? undefined : process.argv[imageIndex + 1] }).then(result => {
if (process.argv.includes('--transcript')) process.stdout.write(result.transcript.map(x => JSON.stringify(x)).join('\n') + '\n');
process.stdout.write(JSON.stringify({ complete: result.complete, error: result.failure,
files: [...result.artifacts].map(([name, data]) => ({ name, bytes: data.length, sha256: createHash('sha256').update(data).digest('hex') })) }) + '\n');
if (!result.complete) process.exitCode = 1;
}).catch(err => { process.stderr.write(String(err) + '\n'); process.exitCode = 1; });
}

38
agent/dev/image-checks.ts Normal file
View File

@ -0,0 +1,38 @@
import { spawn, execFileSync } from 'node:child_process';
import assert from 'node:assert/strict';
import { dockerFlags, runPeer } from './fake-backend.js';
const image = process.argv[2] || 'confluence-pi-agent:rev1';
const security = JSON.parse(execFileSync('docker', ['info', '--format', '{{json .SecurityOptions}}'], { encoding: 'utf8' }));
assert(security.includes('name=rootless'), 'A rootless Docker daemon is required');
console.log('rootless daemon verified');
for (const mode of ['happy', 'happy', 'no-artifacts', 'empty-file', 'skip', 'background', 'isolation', 'isolation']) {
const result = await runPeer({ image, mode });
assert(result.complete, JSON.stringify(result.failure) + result.stderr);
console.log(`PASS image ${mode}`);
}
for (const mode of ['upstream-failure', 'bad-reply', 'duplicate', 'eof-model', 'eof-collection', 'eof-begin', 'eof-end', 'context-limit', 'output-limit', 'stalled-model', 'stalled-collection', 'stop-bridge', 'failure-during-collection']) {
const started = Date.now();
const result = await runPeer({ image, mode, timeoutMs: ['stalled-model', 'stalled-collection', 'stop-bridge'].includes(mode) ? 1500 : 10000 });
assert(!result.complete);
if (mode === 'failure-during-collection') assert.deepEqual(result.transcript.filter(f => f.id.startsWith('a_')).slice(-2).map(f => f.type), ['collection_start', 'error']);
if (['stalled-model', 'stalled-collection', 'stop-bridge'].includes(mode)) assert(Date.now() - started < 5000, 'Independent deadline failed');
console.log(`PASS image failure ${mode}`);
}
// An unchanged 180-second process-start deadline, with no start frame at all.
const name = `pi-no-start-${process.pid}`;
const child = spawn('docker', ['run', '--name', name, ...dockerFlags, image], { stdio: ['pipe', 'pipe', 'pipe'] });
child.stdout.resume(); child.stderr.resume();
const started = Date.now();
const timeout = setTimeout(() => { execFileSync('docker', ['rm', '-f', name]); }, 190000);
console.log('checking unchanged 180-second missing-start deadline');
const progress = setInterval(() => console.log(`missing-start check: ${Math.floor((Date.now() - started) / 1000)}s elapsed`), 30000);
try {
const code = await new Promise(resolve => child.on('exit', resolve));
assert.notEqual(code, 0);
assert(Date.now() - started >= 175000 && Date.now() - started < 185000, 'Missing-start lifetime was not enforced');
console.log('PASS image missing-start lifetime');
} finally {
clearTimeout(timeout); clearInterval(progress); child.stdin.destroy();
spawn('docker', ['rm', '-f', name], { stdio: 'ignore' });
}
console.log(execFileSync('docker', ['image', 'inspect', image, '--format', '{{.Id}}'], { encoding: 'utf8' }).trim());

115
agent/dev/image-packages.txt Normal file
View File

@ -0,0 +1,115 @@
adduser 3.134
apt 2.6.1
base-files 12.4+deb12u15
base-passwd 3.6.1
bash 5.2.15-2+b13
bsdutils 1:2.38.1-5+deb12u3
coreutils 9.1-1
dash 0.5.12-2
debconf 1.5.82
debian-archive-keyring 2023.3+deb12u2
debianutils 5.7-0.5~deb12u1
diffutils 1:3.8-4
dpkg 1.21.23
e2fsprogs 1.47.0-2+b2
findutils 4.9.0-4
gawk 1:5.2.1-2
gcc-12-base 12.2.0-14+deb12u1
gpgv 2.2.40-1.1+deb12u2
grep 3.8-5
gzip 1.12-1
hostname 3.23+nmu1
init-system-helpers 1.65.2+deb12u1
libacl1 2.3.1-3
libapt-pkg6.0 2.6.1
libattr1 1:2.5.1-4
libaudit-common 1:3.0.9-1
libaudit1 1:3.0.9-1
libblkid1 2.38.1-5+deb12u3
libbz2-1.0 1.0.8-5+b1
libc-bin 2.36-9+deb12u14
libc6 2.36-9+deb12u14
libcap-ng0 0.8.3-1+b3
libcap2 1:2.66-4+deb12u3+b1
libcom-err2 1.47.0-2+b2
libcrypt1 1:4.4.33-2
libdb5.3 5.3.28+dfsg2-1
libdebconfclient0 0.270
libexpat1 2.5.0-1+deb12u2
libext2fs2 1.47.0-2+b2
libffi8 3.4.4-1
libgcc-s1 12.2.0-14+deb12u1
libgcrypt20 1.10.1-3+deb12u1
libgmp10 2:6.2.1+dfsg1-1.1
libgnutls30 3.7.9-2+deb12u7
libgpg-error0 1.46-1
libgssapi-krb5-2 1.20.1-2+deb12u5
libhogweed6 3.8.1-2
libidn2-0 2.3.3-1+b1
libk5crypto3 1.20.1-2+deb12u5
libkeyutils1 1.6.3-2
libkrb5-3 1.20.1-2+deb12u5
libkrb5support0 1.20.1-2+deb12u5
liblz4-1 1.9.4-1
liblzma5 5.4.1-1+deb12u1
libmd0 1.0.4-2
libmount1 2.38.1-5+deb12u3
libmpfr6 4.2.0-1
libncursesw6 6.4-4
libnettle8 3.8.1-2
libnsl2 1.3.0-2
libp11-kit0 0.24.1-2
libpam-modules 1.5.2-6+deb12u2
libpam-modules-bin 1.5.2-6+deb12u2
libpam-runtime 1.5.2-6+deb12u2
libpam0g 1.5.2-6+deb12u2
libpcre2-8-0 10.42-1
libproc2-0 2:4.0.2-3
libpython3-stdlib 3.11.2-1+b1
libpython3.11-minimal 3.11.2-6+deb12u8
libpython3.11-stdlib 3.11.2-6+deb12u8
libreadline8 8.2-1.3
libseccomp2 2.5.4-1+deb12u1
libselinux1 3.4-1+b6
libsemanage-common 3.4-1
libsemanage2 3.4-1+b5
libsepol2 3.4-2.1
libsigsegv2 2.14-1
libsmartcols1 2.38.1-5+deb12u3
libsqlite3-0 3.40.1-2+deb12u2
libss2 1.47.0-2+b2
libssl3 3.0.20-1~deb12u2
libstdc++6 12.2.0-14+deb12u1
libsystemd0 252.39-1~deb12u2
libtasn1-6 4.19.0-2+deb12u1
libtinfo6 6.4-4
libtirpc-common 1.3.3+ds-1
libtirpc3 1.3.3+ds-1
libudev1 252.39-1~deb12u2
libunistring2 1.0-2
libuuid1 2.38.1-5+deb12u3
libxxhash0 0.8.1-1
libzstd1 1.5.4+dfsg2-5
login 1:4.13+dfsg1-1+deb12u2
logsave 1.47.0-2+b2
mawk 1.3.4.20200120-3.1
media-types 10.0.0
mount 2.38.1-5+deb12u3
ncurses-base 6.4-4
ncurses-bin 6.4-4
passwd 1:4.13+dfsg1-1+deb12u2
perl-base 5.36.0-7+deb12u3
procps 2:4.0.2-3
python3 3.11.2-1+b1
python3-minimal 3.11.2-1+b1
python3.11 3.11.2-6+deb12u8
python3.11-minimal 3.11.2-6+deb12u8
readline-common 8.2-1.3
sed 4.9-1+deb12u1
sysvinit-utils 3.06-4
tar 1.34+dfsg-1.2+deb12u1
tzdata 2026b-0+deb12u1
usr-is-merged 37~deb12u1
util-linux 2.38.1-5+deb12u3
util-linux-extra 2.38.1-5+deb12u3
zlib1g 1:1.2.13.dfsg-1

78
agent/dev/isolation-probe.py Normal file
View File

@ -0,0 +1,78 @@
# Executed only by the fake model inside the target image; never on the host.
import ctypes
import os
import signal
import time
from pathlib import Path
assert os.getuid() == os.getgid() == 10001
assert Path.cwd() == Path('/work')
assert list(Path('/home/agent').iterdir()) == []
assert {p.name for p in Path('/work').iterdir()} == {'scratch.md', 'artifacts'}
assert set(os.environ) <= {'PATH', 'HOME', 'LANG', 'LC_CTYPE', 'PWD', 'SHLVL', '_'}
assert not Path('/home/vptyp').exists()
assert not Path('/opt/agent/.env').exists()
assert not Path('/opt/agent/node_modules/.env').exists()
assert Path('/sys/fs/cgroup/memory.max').read_text().strip() == '1073741824'
assert Path('/sys/fs/cgroup/pids.max').read_text().strip() == '128'
assert Path('/sys/fs/cgroup/cpu.max').read_text().split() == ['100000', '100000']
for line in Path('/proc/self/status').read_text().splitlines():
if line.startswith('CapEff:'):
assert int(line.split()[1], 16) == 0
if line.startswith('NoNewPrivs:'):
assert line.split()[1] == '1'
if line.startswith('Seccomp:'):
assert line.split()[1] == '2'
for name in ['/opt/agent/supervisor', '/opt/agent/bridge.ts', '/etc/probe-canary']:
try:
with open(name, 'w') as f:
f.write('tamper')
except OSError:
pass
else:
raise AssertionError('Root filesystem is writable')
for name in ['/proc/1/mem', '/proc/1/environ', '/proc/1/fd']:
try:
if name.endswith('/fd'):
os.listdir(name)
else:
open(name, 'rb').close()
except PermissionError:
pass
else:
raise AssertionError('Supervisor proc access is permitted')
assert ctypes.CDLL(None).ptrace(16, 1, 0, 0) == -1 # PTRACE_ATTACH
for sig in [signal.SIGSTOP, signal.SIGKILL, signal.SIGTERM, signal.SIGUSR1]:
os.kill(1, sig)
time.sleep(0.1)
assert Path('/proc/1/stat').read_text().rsplit(')', 1)[1].split()[0] not in ['T', 't', 'Z']
# Actually exhaust the PID cgroup with sleepers; reap them before continuing.
children = []
try:
for i in range(140):
try:
pid = os.fork()
except BlockingIOError:
break
if pid == 0:
time.sleep(10)
os._exit(0)
children.append(pid)
assert 1 < len(children) < 128, 'PID limit was not enforced'
finally:
for pid in children:
os.kill(pid, signal.SIGKILL)
for pid in children:
os.waitpid(pid, 0)
# Exercise memory enforcement rather than only reading the configured limit.
before = Path('/sys/fs/cgroup/memory.events').read_text()
pid = os.fork()
if pid == 0:
allocation = bytearray(1200 * 1024 * 1024)
os._exit(0)
_, status = os.waitpid(pid, 0)
assert os.WIFSIGNALED(status) and os.WTERMSIG(status) == signal.SIGKILL
after = Path('/sys/fs/cgroup/memory.events').read_text()
count = lambda text: int(dict(line.split() for line in text.splitlines())['oom_kill'])
assert count(after) > count(before), 'Memory limit did not produce a cgroup OOM kill'
Path('/home/agent/canary').write_text('must disappear on the next run')
print('isolation checks passed')

10
agent/dev/known-good-summary.json Normal file
View File

@ -0,0 +1,10 @@
{
"complete": true,
"files": [
{
"name": "checklist.md",
"bytes": 32,
"sha256": "0e3fd36684d7b698acd23b70946a1e6c0fb501ee7187a307ce2e1908c504e995"
}
]
}

29
agent/dev/known-good.ndjson Normal file
View File

File diff suppressed because one or more lines are too long

138
agent/framing.ts Normal file
View File

@ -0,0 +1,138 @@
import { Transform, type TransformCallback } from 'node:stream';
import { TextDecoder } from 'node:util';
import { jsonPieces } from './json.js';
import { LIMITS, type ErrorShape, type BridgeBaseMessage } from './types.js';
export function isValidId(id: unknown): id is string {
return typeof id === 'string' && /^[A-Za-z0-9_-]{1,128}$/.test(id);
}
const messages: Record<string, string> = {
invalid_input: 'Invalid runtime protocol or input.',
connectivity_failed: 'The backend connection was lost.',
query_timeout: 'The runtime deadline expired.',
model_context_exceeded: 'The configured model context limit was exceeded.',
model_output_limit: 'The model output is incomplete or exceeds its configured limit.',
upstream_failed: 'The model request failed.',
upstream_response_too_large: 'The upstream response exceeds the supported limit.',
tls_failed: 'The upstream TLS connection failed.',
confluence_auth_failed: 'Confluence authentication failed.',
destination_denied: 'The requested Confluence destination is denied.',
cleanup_failed: 'Runtime cleanup failed.',
execution_failed: 'The runtime could not complete execution.',
};
// Never reflect arbitrary exceptions or peer content into terminal diagnostics.
export function sanitizeError(code: string, _raw?: unknown): ErrorShape {
const safeCode = Object.hasOwn(messages, code) ? code : 'execution_failed';
return { code: safeCode, message: messages[safeCode] };
}
export class NDJsonFrameParser extends Transform {
private parts: Buffer[] = [];
private bytes = 0;
private slab = Buffer.alloc(65536);
private used = 0;
constructor(private maxFrameBytes: number = LIMITS.ORDINARY_FRAME_MAX_BYTES) {
super({ readableObjectMode: true });
}
setMaxFrameBytes(max: number): void { this.maxFrameBytes = max; }
_transform(chunk: Buffer, _encoding: BufferEncoding, callback: TransformCallback): void {
try {
let offset = 0;
while (offset < chunk.length) {
const end = chunk.indexOf(10, offset);
const stop = end < 0 ? chunk.length : end;
const part = chunk.subarray(offset, stop);
if (this.bytes + part.length > this.maxFrameBytes) throw new Error('Frame limit exceeded');
// Copy slices so a small unfinished line cannot retain a large coalesced chunk.
for (let at = 0; at < part.length;) {
const n = Math.min(this.slab.length - this.used, part.length - at);
part.copy(this.slab, this.used, at, at + n); this.used += n; at += n; this.bytes += n;
if (this.used === this.slab.length) { this.parts.push(this.slab); this.slab = Buffer.alloc(65536); this.used = 0; }
}
if (end < 0) break;
if (this.used) this.parts.push(this.slab.subarray(0, this.used));
const line = Buffer.concat(this.parts, this.bytes);
this.slab = Buffer.alloc(65536); this.used = 0;
this.parts = []; this.bytes = 0;
if (!line.length) throw new Error('Empty frame');
const parsed = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(line));
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) throw new Error('Invalid frame');
if (line.length > (parsed.type === 'start' ? LIMITS.START_FRAME_MAX_BYTES : LIMITS.ORDINARY_FRAME_MAX_BYTES)) throw new Error('Frame limit exceeded');
this.push(parsed);
offset = end + 1;
}
callback();
} catch { callback(new Error('Invalid protocol frame')); }
}
_flush(callback: TransformCallback): void {
callback(this.bytes ? new Error('Truncated protocol frame') : undefined);
}
}
type QueuedWrite = { message: BridgeBaseMessage; onStart?: () => void; resolve: () => void; reject: (error: Error) => void };
export class StreamMessageWriter {
private queue: QueuedWrite[] = [];
private active?: QueuedWrite;
private closed = false;
private failure?: Error;
private idle: Array<() => void> = [];
constructor(private stream: NodeJS.WritableStream) {
stream.on('error', () => { this.failure = new Error('Output stream failed'); });
}
write(message: BridgeBaseMessage, onStart?: () => void): Promise<void> {
if (this.closed) return Promise.reject(new Error('Writer closed'));
return this.enqueue(message, onStart);
}
/** Drop frames not yet started; never splice a terminal error into an active line. */
writeTerminal(message: BridgeBaseMessage): Promise<void> {
if (this.closed) return Promise.reject(new Error('Writer closed'));
this.closed = true;
for (const queued of this.queue.splice(0)) queued.reject(new Error('Write cancelled by terminal failure'));
// An already-started complete/error is itself terminal. A failed/truncated
// write must be reported by stream closure, not a second terminal frame.
if (this.active && ['complete', 'error'].includes(this.active.message.type)) {
return Promise.reject(new Error('Terminal frame already started'));
}
return this.enqueue(message);
}
private enqueue(message: BridgeBaseMessage, onStart?: () => void): Promise<void> {
const result = new Promise<void>((resolve, reject) => this.queue.push({ message, onStart, resolve, reject }));
if (!this.active) void this.pump();
return result;
}
private async pump(): Promise<void> {
while (this.queue.length) {
const current = this.queue.shift()!;
this.active = current;
try {
if (this.failure) throw this.failure;
current.onStart?.();
let batch = '';
const flush = async () => {
const data = batch; batch = '';
await new Promise<void>((resolve, reject) => {
this.stream.write(data, 'utf8', (error?: Error | null) => error ? reject(new Error('Output stream failed')) : resolve());
});
};
for (const part of jsonPieces(current.message)) { batch += part; if (batch.length >= 65536) await flush(); }
batch += '\n'; await flush();
current.resolve();
} catch {
this.failure = new Error('Output stream failed');
current.reject(this.failure);
for (const queued of this.queue.splice(0)) queued.reject(this.failure);
this.closed = true;
}
this.active = undefined;
}
for (const resolve of this.idle.splice(0)) resolve();
}
async close(): Promise<void> {
this.closed = true;
if (this.active || this.queue.length) await new Promise<void>(resolve => this.idle.push(resolve));
if (this.failure) throw this.failure;
}
}
export function logDiagnostic(message: string): void {
process.stderr.write(message.slice(0, 1024) + '\n');
}

30
agent/json.ts Normal file
View File

@ -0,0 +1,30 @@
/** Bounded serialization fragments: large escaped strings never form one copy. */
export function* jsonPieces(value: any, depth = 0): Generator<string> {
if (depth > 64) throw new Error('JSON nesting limit');
if (typeof value === 'string') {
yield '"';
for (let offset = 0; offset < value.length;) {
let end = Math.min(value.length, offset + 16384);
if (end < value.length && /[\uD800-\uDBFF]/.test(value[end - 1])) end--;
yield JSON.stringify(value.slice(offset, end)).slice(1, -1); offset = end;
}
yield '"';
} else if (Array.isArray(value)) {
yield '[';
for (let i = 0; i < value.length; i++) { if (i) yield ','; yield* jsonPieces(value[i] ?? null, depth + 1); }
yield ']';
} else if (value && typeof value === 'object') {
yield '{'; let first = true;
for (const key of Object.keys(value)) {
if (value[key] === undefined || typeof value[key] === 'function' || typeof value[key] === 'symbol') continue;
if (!first) yield ','; first = false;
yield JSON.stringify(key); yield ':'; yield* jsonPieces(value[key], depth + 1);
}
yield '}';
} else { yield JSON.stringify(value) ?? 'null'; }
}
export function jsonBytes(value: any, limit = Number.MAX_SAFE_INTEGER): number {
let bytes = 0;
for (const part of jsonPieces(value)) { bytes += Buffer.byteLength(part); if (bytes > limit) return bytes; }
return bytes;
}

56
agent/local-tools.ts Normal file
View File

@ -0,0 +1,56 @@
import fs from 'node:fs/promises';
import type { AgentTool } from '@earendil-works/pi-agent-core';
import { constants } from 'node:fs';
import { spawn } from 'node:child_process';
import { createCodingTools, type BashOperations } from '@earendil-works/pi-coding-agent';
const maxFile = 16 * 1024 * 1024;
async function readFile(name: string): Promise<Buffer> {
const file = await fs.open(name, constants.O_RDONLY | constants.O_NONBLOCK);
try {
const stat = await file.stat();
if (!stat.isFile() || stat.size > maxFile) throw new Error('Local read requires a regular file of at most 16 MiB; use shell ranges for large files.');
// Bound reads even if another local process grows the file concurrently.
const buffer = Buffer.alloc(stat.size + 1);
let offset = 0;
while (offset < buffer.length) { const { bytesRead } = await file.read(buffer, offset, buffer.length - offset, offset); if (!bytesRead) break; offset += bytesRead; }
if (offset > stat.size) throw new Error('Local file changed while reading.');
return buffer.subarray(0, offset);
} finally { await file.close(); }
}
const operations: BashOperations = {
exec: (command, cwd, options) => new Promise((resolve, reject) => {
const child = spawn('/bin/bash', ['--noprofile', '--norc', '-c', command], { cwd, detached: true, env: { PATH: process.env.PATH, HOME: process.env.HOME, LANG: 'C.UTF-8' }, stdio: ['ignore', 'pipe', 'pipe'] });
let bytes = 0, capped = false, timedOut = false;
const kill = () => { if (child.pid) { try { process.kill(-child.pid, 'SIGKILL'); } catch {} } };
const output = (data: Buffer) => {
if (capped) return;
const remaining = 1024 * 1024 - bytes;
options.onData(data.subarray(0, remaining)); bytes += Math.min(data.length, remaining);
if (data.length > remaining) { capped = true; options.onData(Buffer.from('\n[Local shell output capped at 1 MiB.]\n')); kill(); }
};
child.stdout.on('data', output); child.stderr.on('data', output);
options.signal?.addEventListener('abort', kill, { once: true });
if (options.signal?.aborted) kill();
const timer = setTimeout(() => { timedOut = true; kill(); }, Math.min(30000, (options.timeout ?? 30) * 1000));
child.on('error', () => reject(new Error('Local shell could not start.')));
child.on('close', code => {
clearTimeout(timer); options.signal?.removeEventListener('abort', kill);
if (options.signal?.aborted) reject(new Error('aborted'));
else if (timedOut) reject(new Error(`timeout:${Math.min(30, options.timeout ?? 30)}`));
else resolve({ exitCode: capped ? 1 : code });
});
}),
};
export function localTools(work: string): AgentTool<any>[] {
const tools = createCodingTools(work, {
bash: { operations, exposeSessionEnvironment: false, shellPath: '/bin/bash' },
read: { operations: { readFile, access: p => fs.access(p, constants.R_OK), detectImageMimeType: async () => null } },
edit: { operations: { readFile, access: p => fs.access(p, constants.R_OK | constants.W_OK), writeFile: async (p, content) => {
if (Buffer.byteLength(content) > maxFile) throw new Error('Local edit result exceeds 16 MiB.');
await fs.writeFile(p, content);
} } },
});
const read = tools.find(t => t.name === 'read')!;
read.description = 'Read regular text files up to 16 MiB. Output is truncated to 2000 lines or 50 KiB; use offset/limit for continuation. Use shell ranges for larger files. Image attachments are not supported by this text/tool model adapter.';
return tools;
}

277
agent/model-provider.ts Normal file
View File

@ -0,0 +1,277 @@
import type {
Context,
Model,
AssistantMessage,
SimpleStreamOptions,
TextContent,
ToolCall
} from "@earendil-works/pi-ai";
import { createAssistantMessageEventStream } from "@earendil-works/pi-ai";
import type { StreamFn } from "@earendil-works/pi-agent-core";
import type {
ModelDescriptor,
ModelRequest,
ModelResponse,
MessageContract,
ContentBlock,
ToolDeclaration
} from "./types.js";
/**
* Converts pi-ai Context into CONTRACTS ModelRequest format.
* Invariants:
* - System instruction is NOT duplicated into messages.
* - Tool calls use type: "tool_call".
* - Tool results use role: "tool", tool_call_id, is_error.
* - Provider state is round-tripped on assistant messages.
*/
export function contextToModelRequest(context: Context): ModelRequest {
const messages: MessageContract[] = [];
for (const m of context.messages) {
if (m.role === "user") {
let textContent: string = "";
if (typeof m.content === "string") {
textContent = m.content;
} else if (Array.isArray(m.content)) {
textContent = m.content
.filter((c: any) => c.type === "text")
.map((c: any) => c.text)
.join("\n");
}
messages.push({
role: "user",
content: [{ type: "text", text: textContent }]
});
} else if (m.role === "assistant") {
const contentBlocks: ContentBlock[] = [];
for (const block of m.content) {
if (block.type === "text") {
contentBlocks.push({
type: "text",
text: block.text
});
} else if (block.type === "toolCall") {
contentBlocks.push({
type: "tool_call",
id: block.id,
name: block.name,
arguments: block.arguments || {}
});
}
// Thinking blocks are not part of shared model boundary
}
const assistantMsg: MessageContract = {
role: "assistant",
content: contentBlocks,
...((m as any).provider_state !== undefined ? { provider_state: (m as any).provider_state } : {})
};
messages.push(assistantMsg);
} else if (m.role === "toolResult") {
let text = "";
if (typeof m.content === "string") {
text = m.content;
} else if (Array.isArray(m.content)) {
text = m.content
.filter((c: any) => c.type === "text")
.map((c: any) => c.text)
.join("\n");
}
messages.push({
role: "tool",
tool_call_id: m.toolCallId,
name: m.toolName,
content: text,
is_error: Boolean(m.isError)
});
}
}
const tools: ToolDeclaration[] = (context.tools || []).map((t) => ({
name: t.name,
description: t.description || "",
input_schema: (t.parameters as any) || {}
}));
return {
messages,
tools
};
}
/**
* Converts CONTRACTS ModelResponse to pi-ai AssistantMessage.
*/
export function modelResponseToAssistantMessage(
response: ModelResponse,
model: Model<any>
): AssistantMessage {
const content: (TextContent | ToolCall)[] = [];
for (const block of response.content) {
if (block.type === "text") {
content.push({
type: "text",
text: block.text
});
} else if (block.type === "tool_call") {
content.push({
type: "toolCall",
id: block.id,
name: block.name,
arguments: block.arguments || {}
});
}
}
let stopReason: AssistantMessage["stopReason"] = "stop";
if (response.stop_reason === "tool_calls") {
stopReason = "toolUse";
} else if (response.stop_reason === "length") {
stopReason = "length";
} else if (response.stop_reason === "stop") {
stopReason = "stop";
}
const assistantMessage: AssistantMessage = {
role: "assistant",
content,
api: model.api,
provider: model.provider,
model: model.id,
usage: {
input: response.usage?.input_tokens ?? 0,
output: response.usage?.output_tokens ?? 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: (response.usage?.input_tokens ?? 0) + (response.usage?.output_tokens ?? 0),
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }
},
stopReason,
timestamp: Date.now()
};
if (response.provider_state !== undefined) {
(assistantMessage as any).provider_state = response.provider_state;
}
return assistantMessage;
}
export type SendModelRequestFn = (request: ModelRequest) => Promise<ModelResponse>;
/**
* Creates a Model definition and StreamFn that routes model calls across the bridge.
*/
export function createModelProvider(
descriptor: ModelDescriptor,
sendModelRequest: SendModelRequestFn
): { model: Model<any>; streamFn: StreamFn } {
const model: Model<any> = {
id: descriptor.id,
name: descriptor.id,
api: "custom" as any,
provider: "backend",
baseUrl: "",
reasoning: false,
input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: descriptor.context_window_tokens,
maxTokens: descriptor.max_output_tokens
};
const streamFn: StreamFn = async (_model, context, options) => {
const stream = createAssistantMessageEventStream();
(async () => {
try {
if (options?.signal?.aborted) {
const abortedMsg: AssistantMessage = {
role: "assistant",
content: [],
api: model.api,
provider: model.provider,
model: model.id,
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
stopReason: "aborted",
errorMessage: "Request aborted",
timestamp: Date.now()
};
stream.push({ type: "error", reason: "aborted", error: abortedMsg });
stream.end(abortedMsg);
return;
}
const modelRequest = contextToModelRequest(context);
const response = await sendModelRequest(modelRequest);
const assistantMessage = modelResponseToAssistantMessage(response, model);
if (options?.signal?.aborted) {
const abortedMsg: AssistantMessage = {
role: "assistant",
content: [],
api: model.api,
provider: model.provider,
model: model.id,
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
stopReason: "aborted",
errorMessage: "Request aborted",
timestamp: Date.now()
};
stream.push({ type: "error", reason: "aborted", error: abortedMsg });
stream.end(abortedMsg);
return;
}
// Emit streaming events
const partial: AssistantMessage = {
...assistantMessage,
content: [],
stopReason: "pending" as any
};
stream.push({ type: "start", partial: { ...partial } });
for (let i = 0; i < assistantMessage.content.length; i++) {
const block = assistantMessage.content[i];
if (block.type === "text") {
partial.content = [...partial.content, { type: "text", text: "" }];
stream.push({ type: "text_start", contentIndex: i, partial: { ...partial } });
partial.content[i] = { type: "text", text: block.text };
stream.push({ type: "text_delta", contentIndex: i, delta: block.text, partial: { ...partial } });
stream.push({ type: "text_end", contentIndex: i, content: block.text, partial: { ...partial } });
} else if (block.type === "toolCall") {
partial.content = [...partial.content, { type: "toolCall", id: block.id, name: block.name, arguments: {} }];
stream.push({ type: "toolcall_start", contentIndex: i, partial: { ...partial } });
partial.content[i] = block;
stream.push({ type: "toolcall_delta", contentIndex: i, delta: JSON.stringify(block.arguments), partial: { ...partial } });
stream.push({ type: "toolcall_end", contentIndex: i, toolCall: block, partial: { ...partial } });
}
}
const validDoneReason = assistantMessage.stopReason as "stop" | "toolUse" | "length";
stream.push({ type: "done", reason: validDoneReason, message: assistantMessage });
stream.end(assistantMessage);
} catch (err: any) {
const errorMsg: AssistantMessage = {
role: "assistant",
content: [],
api: model.api,
provider: model.provider,
model: model.id,
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
stopReason: "error",
errorMessage: "Model request failed.",
timestamp: Date.now()
};
stream.push({ type: "error", reason: "error", error: errorMsg });
stream.end(errorMsg);
}
})();
return stream;
};
return { model, streamFn };
}

3993
agent/package-lock.json generated Normal file
View File

File diff suppressed because it is too large Load Diff

24
agent/package.json Normal file
View File

@ -0,0 +1,24 @@
{
"name": "confluence-agent-runtime",
"version": "1.0.0",
"description": "Isolated pi runtime and bridge for Confluence research",
"type": "module",
"scripts": {
"build": "tsc",
"test": "node --import tsx --test tests/*.test.ts",
"start": "python3 supervisor",
"proof": "npm run build && node dist/dev/fake-backend.js",
"test:image": "npm run build && node dist/dev/image-checks.js"
},
"dependencies": {
"@earendil-works/pi-coding-agent": "0.85.1",
"@earendil-works/pi-agent-core": "0.85.1",
"@earendil-works/pi-ai": "0.85.1",
"typebox": "1.3.7"
},
"devDependencies": {
"@types/node": "22.20.2",
"tsx": "4.23.13",
"typescript": "5.9.3"
}
}

22
agent/supervision.ts Normal file
View File

@ -0,0 +1,22 @@
import net from 'node:net';
import { createInterface } from 'node:readline';
export function supervisorChannel(): { deadline: (ms: number) => void; reap: () => Promise<void> } | undefined {
const fd = process.env.AGENT_SUPERVISOR_FD;
if (!fd || !/^\d+$/.test(fd)) return undefined;
const socket = new net.Socket({ fd: Number(fd), readable: true, writable: true });
const lines = createInterface({ input: socket });
let awaiting: { resolve: () => void; reject: (err: Error) => void } | undefined;
lines.on('line', line => {
if (line !== '{"type":"reaped"}' || !awaiting) { socket.destroy(); return; }
awaiting.resolve(); awaiting = undefined;
});
socket.on('error', () => { awaiting?.reject(new Error('Supervisor unavailable')); awaiting = undefined; });
socket.on('close', () => { awaiting?.reject(new Error('Supervisor unavailable')); awaiting = undefined; });
return {
deadline: ms => { socket.write(JSON.stringify({ type: 'set_deadline', remaining_ms: ms }) + '\n'); },
reap: () => new Promise<void>((resolve, reject) => {
awaiting = { resolve, reject };
socket.write('{"type":"reap_descendants"}\n');
}),
};
}

143
agent/supervisor Executable file
View File

@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""Immutable Linux PID-1 supervisor; host mode is development-only."""
import ctypes
import json
import os
import select
import signal
import socket
import subprocess
import sys
import time
from pathlib import Path
libc = ctypes.CDLL(None, use_errno=True)
# Adopt double-forked descendants on hosts, as well as under container PID 1.
if libc.prctl(36, 1, 0, 0, 0) != 0: # PR_SET_CHILD_SUBREAPER
sys.exit(1)
if libc.prctl(4, 0, 0, 0, 0) != 0: # PR_SET_DUMPABLE: deny same-UID proc memory/fd access
sys.exit(1)
container = os.getpid() == 1
if not container and '--dev' not in sys.argv:
sys.stderr.write('Supervisor requires container PID 1; use --dev only for host tests.\n')
sys.exit(1)
# PID 1 ignores namespace-local SIGKILL/SIGSTOP. Other signals terminate the
# run or are ignored; none can pause or extend the deadline. No exec after prctl.
for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP, signal.SIGUSR1, signal.SIGUSR2):
signal.signal(sig, signal.SIG_IGN)
started = time.monotonic()
deadline = started + 180
parent, child = socket.socketpair()
child.set_inheritable(True)
root = Path(__file__).resolve().parent
entry = root / 'dist' / 'bridge.js'
args = ['node', str(entry)]
env = {'PATH': '/usr/local/bin:/usr/bin:/bin', 'HOME': '/home/agent', 'LANG': 'C.UTF-8',
'AGENT_SUPERVISOR_FD': str(child.fileno())}
if '--dev' in sys.argv:
env['PATH'] = os.environ.get('PATH', env['PATH'])
env['HOME'] = os.environ.get('HOME', '/tmp')
env['AGENT_WORK_DIR'] = os.environ.get('AGENT_WORK_DIR', '/work')
bridge = subprocess.Popen(args, env=env, pass_fds=(child.fileno(),))
child.close()
parent.setblocking(False)
protected = {os.getpid(), bridge.pid}
buffer = b''
def descendants():
"""Never signal unrelated host processes. Adopted children retain this ancestry."""
parents = {}
for item in Path('/proc').iterdir():
if item.name.isdecimal():
try:
fields = (item / 'stat').read_text().rsplit(')', 1)[1].split()
parents[int(item.name)] = int(fields[1])
except (OSError, ValueError, IndexError):
pass
owned = {os.getpid()}
while True:
extra = {pid for pid, ppid in parents.items() if ppid in owned} - owned
if not extra:
break
owned.update(extra)
return owned - protected
def reap():
# Re-scan until every descendant is gone, including children forked during
# termination. Reap only adopted children, leaving Popen to reap the bridge.
while time.monotonic() < deadline:
victims = descendants()
if not victims:
return
for pid in victims:
try:
os.kill(pid, signal.SIGKILL)
except ProcessLookupError:
pass
try:
os.waitpid(pid, os.WNOHANG)
except ChildProcessError:
pass
time.sleep(0.01)
raise TimeoutError()
code = 1
try:
deadline_set = False
collecting = False
while time.monotonic() < deadline:
if bridge.poll() is not None:
code = bridge.returncode
break
readable, _, _ = select.select([parent], [], [], min(0.05, max(0, deadline - time.monotonic())))
if not readable:
continue
data = parent.recv(4096)
if not data:
try:
code = bridge.wait(timeout=min(1, max(0.01, deadline - time.monotonic())))
except subprocess.TimeoutExpired:
pass
break
buffer += data
if len(buffer) > 8192:
break
while b'\n' in buffer:
line, buffer = buffer.split(b'\n', 1)
msg = json.loads(line)
if not isinstance(msg, dict):
raise ValueError()
if msg.get('type') == 'set_deadline' and not deadline_set:
ms = msg.get('remaining_ms')
if type(ms) is not int or not 1 <= ms <= 180000:
raise ValueError()
deadline = min(deadline, time.monotonic() + ms / 1000)
deadline_set = True
elif msg == {'type': 'reap_descendants'} and not collecting:
reap()
collecting = True
parent.sendall(b'{"type":"reaped"}\n')
else:
raise ValueError()
except (OSError, ValueError, TimeoutError):
pass
finally:
if container and time.monotonic() >= deadline:
sys.exit(1) # Kernel destroys the namespace; do not add cleanup time.
# Killing PID 1 ends the container namespace. Host development cleanup is
# still restricted to descendants of this supervisor.
protected.discard(bridge.pid)
deadline = max(deadline, time.monotonic() + 1)
try:
reap()
except (OSError, TimeoutError):
pass
try:
bridge.wait(timeout=1)
except subprocess.TimeoutExpired:
bridge.kill()
parent.close()
sys.exit(code if code >= 0 else 1)

91
agent/tests/artifacts.test.ts Normal file
View File

@ -0,0 +1,91 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { execFileSync } from 'node:child_process';
import { scanArtifactsDirectory, exportArtifacts, closeArtifacts, validateRelativeArtifactPath } from '../artifacts.js';
function fixture(t: any) {
const root = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-artifacts-'));
t.after(() => fs.rmSync(root, { recursive: true, force: true }));
return root;
}
test('path validation rejects traversal and portable-name violations', () => {
for (const name of ['', '/x', '../x', 'a/./x', 'a//x', 'a\\x', 'a\n', 'x'.repeat(1025)]) assert(!validateRelativeArtifactPath(name).valid);
assert(validateRelativeArtifactPath('docs/雪.md').valid);
});
test('symlink parents, hardlinks, FIFO and normalized duplicates are omitted', t => {
const root = fixture(t); const exports = path.join(root, 'artifacts'); fs.mkdirSync(exports);
fs.writeFileSync(path.join(exports, 'ok'), 'bytes');
fs.writeFileSync(path.join(root, 'private'), 'private');
fs.symlinkSync(root, path.join(exports, 'parent'));
fs.symlinkSync(path.join(root, 'private'), path.join(exports, 'link'));
fs.linkSync(path.join(root, 'private'), path.join(exports, 'hard'));
execFileSync('mkfifo', [path.join(exports, 'fifo')]);
fs.writeFileSync(path.join(exports, 'é'), '1'); fs.writeFileSync(path.join(exports, 'e\u0301'), '2');
const scan = scanArtifactsDirectory(exports); t.after(() => closeArtifacts(scan.candidates));
assert.equal(scan.candidates.length, 2); assert(scan.warnings.some(w => w.code === 'artifact_not_regular' && /Repeated 4 times/.test(w.message)));
assert(scan.warnings.some(w => w.code === 'artifact_duplicate_name'));
fs.symlinkSync(exports, path.join(root, 'alias'));
assert.equal(scanArtifactsDirectory(path.join(root, 'alias')).candidates.length, 0);
});
test('held descriptor prevents a replaced parent from redirecting export', async t => {
const root = fixture(t); const dir = path.join(root, 'artifacts'); fs.mkdirSync(dir); fs.mkdirSync(path.join(dir, 'nested'));
fs.writeFileSync(path.join(dir, 'nested', 'ok'), 'safe');
const scan = scanArtifactsDirectory(dir); t.after(() => closeArtifacts(scan.candidates));
fs.renameSync(path.join(dir, 'nested'), path.join(root, 'moved'));
fs.symlinkSync(root, path.join(dir, 'nested'));
fs.writeFileSync(path.join(root, 'ok'), 'evil');
let bytes = '';
assert.equal(await exportArtifacts(scan.candidates, { sendBegin: async () => 'accept', sendChunk: async (_id, _i, b) => { bytes += Buffer.from(b, 'base64').toString(); }, sendEnd: async () => {} }), 1);
assert.equal(bytes, 'safe');
});
test('empty files, skip and bounded chunks; changed files fail', async t => {
const root = fixture(t); fs.writeFileSync(path.join(root, 'empty'), ''); fs.writeFileSync(path.join(root, 'data'), Buffer.alloc(70000));
const scan = scanArtifactsDirectory(root); t.after(() => closeArtifacts(scan.candidates));
const chunks: number[] = []; const ends: number[] = [];
assert.equal(await exportArtifacts(scan.candidates, { sendBegin: async () => 'accept', sendChunk: async (_id, _i, b) => { chunks.push(Buffer.from(b, 'base64').length); }, sendEnd: async (_id, _size, n) => { ends.push(n); } }), 2);
assert.deepEqual(chunks, [65536, 4464]); assert(ends.includes(0));
assert.equal(await exportArtifacts(scan.candidates, { sendBegin: async () => 'skip', sendChunk: async () => { throw Error(); }, sendEnd: async () => { throw Error(); } }), 0);
fs.appendFileSync(path.join(root, 'data'), 'x');
await assert.rejects(exportArtifacts(scan.candidates, { sendBegin: async () => 'accept', sendChunk: async () => {}, sendEnd: async () => {} }));
});
test('file count, individual size and total byte limits', t => {
const root = fixture(t);
for (let i = 0; i < 25; i++) fs.writeFileSync(path.join(root, `f${i}`), '');
fs.closeSync(fs.openSync(path.join(root, 'oversized'), 'w')); fs.truncateSync(path.join(root, 'oversized'), 10 * 1024 * 1024 + 1);
const scan = scanArtifactsDirectory(root); t.after(() => closeArtifacts(scan.candidates));
assert.equal(scan.candidates.length, 20); assert(scan.warnings.some(w => w.code === 'artifact_oversized'));
const large = fixture(t);
for (let i = 0; i < 6; i++) { const p = path.join(large, `f${i}`); fs.writeFileSync(p, ''); fs.truncateSync(p, 10 * 1024 * 1024); }
const largeScan = scanArtifactsDirectory(large); t.after(() => closeArtifacts(largeScan.candidates));
assert.equal(largeScan.candidates.length, 5); assert(largeScan.warnings.some(w => w.code === 'artifact_total_limit_exceeded'));
});
test('socket files are omitted', async t => {
const root = fixture(t);
const { createServer } = await import('node:net');
const server = createServer();
await new Promise<void>((resolve, reject) => { server.on('error', reject); server.listen(path.join(root, 'socket'), resolve); });
t.after(() => new Promise<void>(resolve => server.close(() => resolve())));
const scan = scanArtifactsDirectory(root); t.after(() => closeArtifacts(scan.candidates));
assert.equal(scan.candidates.length, 0); assert(scan.warnings.some(w => w.code === 'artifact_not_regular'));
});
test('a growing file fails without sending excess bytes', async t => {
const root = fixture(t); const file = path.join(root, 'growing'); fs.writeFileSync(file, Buffer.alloc(70000));
const scan = scanArtifactsDirectory(root); t.after(() => closeArtifacts(scan.candidates));
let sent = 0;
await assert.rejects(exportArtifacts(scan.candidates, { sendBegin: async () => 'accept', sendChunk: async (_id, _index, data) => {
sent += Buffer.from(data, 'base64').length; fs.appendFileSync(file, Buffer.alloc(1000));
}, sendEnd: async () => { throw new Error('Changed file must not close a successful transfer'); } }));
assert.equal(sent, 70000);
});
test('repeated filesystem rejections are aggregated without losing counts', t => {
const root = fixture(t);
for (let i = 0; i < 130; i++) fs.symlinkSync('/nonexistent-dummy-target', path.join(root, `link_${i}`));
const scan = scanArtifactsDirectory(root); t.after(() => closeArtifacts(scan.candidates));
assert.equal(scan.candidates.length, 0); assert.equal(scan.warnings.length, 1);
assert.equal(scan.warnings[0].code, 'artifact_not_regular'); assert.match(scan.warnings[0].message, /Repeated 130 times/);
});

220
agent/tests/bridge.test.ts Normal file
View File

@ -0,0 +1,220 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { PassThrough, Writable } from 'node:stream';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { Bridge } from '../bridge.js';
const start = { prompt: 'dummy', system_instruction: 'trusted', remaining_ms: 1000, model: { id: 'dummy', context_window_tokens: 10000, max_output_tokens: 1000 } };
function harness(t: any, script: (f: any, send: (type: string, payload: any, reply?: string) => void) => void, remaining = 1000) {
const work = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-bridge-')); t.after(() => fs.rmSync(work, { recursive: true, force: true }));
const input = new PassThrough(), output = new PassThrough();
const bridge = new Bridge({ stdin: input, stdout: output, workDir: work, onChildReap: async () => {} });
const frames: any[] = []; let bytes = '', seq = 0;
const send = (type: string, payload: any, reply_to?: string) => input.write(JSON.stringify({ v: 1, type, id: `b_${++seq}`, ...(reply_to ? { reply_to } : {}), payload }) + '\n');
output.on('data', part => {
bytes += part.toString(); let end;
while ((end = bytes.indexOf('\n')) >= 0) { const f = JSON.parse(bytes.slice(0, end)); bytes = bytes.slice(end + 1); frames.push(f); script(f, send); }
});
const done = bridge.start(); send('start', { ...start, remaining_ms: remaining });
return { input, bridge, frames, done, work };
}
const final = { content: [{ type: 'text', text: 'answer' }], stop_reason: 'stop', usage: { input_tokens: 2, output_tokens: 2 } };
test('zero artifacts completes only after collection_ready and stays alive until EOF', async t => {
let complete!: () => void; const observed = new Promise<void>(r => complete = r);
const h = harness(t, (f, send) => {
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'complete') complete();
});
await observed; await new Promise<void>(resolve => setImmediate(resolve)); assert.equal(h.bridge.getState(), 'COMPLETE');
let settled = false; h.done.then(() => settled = true);
await new Promise(r => setTimeout(r, 10)); assert(!settled);
h.input.end(); await h.done;
assert.equal(h.frames.at(-1).payload.accepted_transfer_count, 0);
});
for (const mode of ['wrong-reply', 'wrong-type', 'duplicate', 'extra-field', 'eof', 'upstream', 'length', 'empty-answer']) {
test(`terminal model/protocol failure: ${mode}`, async t => {
const h = harness(t, (f, send) => {
if (f.type !== 'model_request') return;
if (mode === 'eof') { h.input.end(); return; }
const result = mode === 'length' ? { ...final, stop_reason: 'length' } : mode === 'empty-answer' ? { ...final, content: [] } : final;
const payload = mode === 'upstream' ? { result: null, error: { code: 'upstream_failed', message: 'SECRET_PAYLOAD' } } : { result, error: null };
send(mode === 'wrong-type' ? 'tool_response' : 'model_response', mode === 'extra-field' ? { ...payload, bad: 1 } : payload, mode === 'wrong-reply' ? 'a_unknown' : f.id);
if (mode === 'duplicate') send('model_response', payload, f.id);
});
await assert.rejects(h.done); assert.equal(h.bridge.getState(), 'FAILED');
await new Promise(r => setTimeout(r, 5));
assert(!h.frames.some(f => f.type === 'complete'));
assert(!JSON.stringify(h.frames.filter(f => f.type === 'error')).includes('SECRET'));
});
}
test('collection correlation and stalled collection retain deadline', async t => {
for (const wrong of [true, false]) {
const h = harness(t, (f, send) => {
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (wrong && f.type === 'collection_start') send('collection_ready', {}, 'a_unknown');
}, 200);
await assert.rejects(h.done);
assert.equal(h.bridge.getState(), 'FAILED');
}
});
test('artifact begin/end acknowledgements must match request, transfer and phase', async t => {
for (const phase of ['begin', 'end']) {
const h = harness(t, (f, send) => {
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'artifact_begin') send('artifact_ack', { transfer_id: f.payload.transfer_id, decision: phase === 'begin' ? 'stored' : 'accept', warning: null }, f.id);
if (f.type === 'artifact_end') send('artifact_ack', { transfer_id: 'a_wrong', decision: 'stored', warning: null }, f.id);
});
fs.mkdirSync(path.join(h.work, 'artifacts'), { recursive: true }); fs.writeFileSync(path.join(h.work, 'artifacts', 'empty'), '');
await assert.rejects(h.done); assert(!h.frames.some(f => f.type === 'complete'));
}
});
test('concurrent remote calls correlate independently and stay bounded', async t => {
let modelFrame: any, sendReply!: (type: string, payload: any, reply?: string) => void;
let modelReady!: () => void; const ready = new Promise<void>(resolve => modelReady = resolve);
const h = harness(t, (f, send) => {
sendReply = send;
if (f.type === 'model_request') { modelFrame = f; modelReady(); }
if (f.type === 'tool_request') send('tool_response', { result: { page_id: f.payload.parameters.page_id, title: '', space: '', url: '', markdown: '', truncated: false }, error: null }, f.id);
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'complete') h.input.end();
});
await ready;
const calls = ['1', '2', '3'].map(page_id => h.bridge.sendToolRequest('confluence_view', { page_id }));
await assert.rejects(h.bridge.sendToolRequest('confluence_view', { page_id: '4' }), /limit/);
assert.deepEqual((await Promise.all(calls)).map(x => x.page_id), ['1', '2', '3']);
sendReply('model_response', { result: final, error: null }, modelFrame.id);
await h.done;
assert.equal(h.frames.filter(f => f.type === 'tool_request').length, 3);
});
test('model call total cannot exceed fifty turns', async t => {
let calls = 0;
const h = harness(t, (f, send) => {
if (f.type !== 'model_request') return;
calls++;
send('model_response', { result: { content: [{ type: 'tool_call', id: `sdk_${calls}`, name: 'read', arguments: { path: '/nonexistent-pi-fixture' } }], stop_reason: 'tool_calls', usage: { input_tokens: 1, output_tokens: 1 } }, error: null }, f.id);
}, 2000);
await assert.rejects(h.done); assert.equal(calls, 50); assert(!h.frames.some(f => f.type === 'complete'));
});
for (const delayed of [false, true]) {
test(`accepted complete stays terminal beyond remaining_ms, delayed model=${delayed}`, async t => {
let observed!: () => void; const complete = new Promise<void>(resolve => observed = resolve);
const h = harness(t, (f, send) => {
if (f.type === 'model_request') {
const respond = () => send('model_response', { result: final, error: null }, f.id);
if (delayed) setTimeout(respond, 200); else respond();
}
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'complete') observed();
}, 400);
let settled = false; let failure: unknown;
h.done.then(() => settled = true, error => failure = error);
await complete;
await new Promise(resolve => setTimeout(resolve, 500));
h.bridge.failRun('query_timeout');
assert.equal(h.bridge.getState(), 'COMPLETE'); assert.equal(failure, undefined); assert(!settled);
assert.deepEqual(h.frames.map(f => f.type), ['model_request', 'collection_start', 'complete']);
h.input.end(); await h.done;
});
}
test('stalled model still expires before completion', async t => {
const h = harness(t, () => {}, 80);
await assert.rejects(h.done, /deadline expired/);
assert.equal(h.bridge.getState(), 'FAILED'); assert(await h.bridge.flushOutput());
assert.equal(h.frames.at(-1).type, 'error');
});
test('synchronous EOF cannot turn a failed complete write into success', async t => {
const work = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-complete-write-'));
t.after(() => fs.rmSync(work, { recursive: true, force: true }));
const input = new PassThrough(); let seq = 0; const frames: any[] = [];
const send = (type: string, payload: any, reply_to: string) => input.write(JSON.stringify({ v: 1, type, id: `b_${++seq}`, reply_to, payload }) + '\n');
const output = new Writable({ write(chunk, _encoding, callback) {
const f = JSON.parse(chunk.toString()); frames.push(f);
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'complete') { input.end(); setTimeout(() => callback(new Error('DUMMY_PRIVATE_STREAM_ERROR')), 20); }
else callback();
} });
const bridge = new Bridge({ stdin: input, stdout: output, workDir: work, onChildReap: async () => {} });
const done = bridge.start(); input.write(JSON.stringify({ v: 1, type: 'start', id: `b_${++seq}`, payload: start }) + '\n');
await assert.rejects(done); assert.equal(bridge.getState(), 'FAILED');
assert.equal(await bridge.flushOutput(), false);
assert.deepEqual(frames.map(f => f.type), ['model_request', 'collection_start', 'complete']);
});
test('failure flushing waits for slow callbacks and stays bounded if output stalls', async () => {
for (const stalled of [false, true]) {
let release!: () => void; const chunks: string[] = [];
const output = new Writable({ write(chunk, _encoding, callback) {
chunks.push(chunk.toString()); release = () => callback();
if (!stalled) setTimeout(release, 60);
} });
const input = new PassThrough(); const bridge = new Bridge({ stdin: input, stdout: output });
const done = bridge.start(); bridge.failRun('invalid_input'); await assert.rejects(done);
const started = Date.now(); const flushed = await bridge.flushOutput(stalled ? 40 : 500);
assert.equal(flushed, !stalled); assert(Date.now() - started < 600);
if (!stalled) assert(Date.now() - started >= 40);
assert.equal(JSON.parse(chunks.join('')).type, 'error');
if (stalled) release(); input.end();
}
});
test('collection warnings retain repeated and overflow counts on the wire', async t => {
let captured: any[] = [];
const h = harness(t, (f, send) => {
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (f.type === 'collection_start') { captured = f.payload.warnings; send('collection_ready', {}, f.id); }
if (f.type === 'complete') h.input.end();
});
for (let i = 0; i < 105; i++) h.bridge.addWarning({ code: `warning_${i}`, message: 'Bounded warning.' });
h.bridge.addWarning({ code: 'warning_0', message: 'Bounded warning.' });
await h.done; assert.equal(captured.length, 100);
assert.match(captured[0].message, /Repeated 2 times/);
assert.equal(captured.at(-1).code, 'warnings_aggregated'); assert.match(captured.at(-1).message, /^6 additional/);
});
test('deadline remains active until the complete write succeeds', async t => {
const work = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-complete-stall-'));
t.after(() => fs.rmSync(work, { recursive: true, force: true }));
const input = new PassThrough(); let seq = 0, release!: () => void; const frames: any[] = [];
const send = (type: string, payload: any, reply_to: string) => input.write(JSON.stringify({ v: 1, type, id: `b_${++seq}`, reply_to, payload }) + '\n');
const output = new Writable({ write(chunk, _encoding, callback) {
const f = JSON.parse(chunk.toString()); frames.push(f);
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'complete') release = () => callback(); else callback();
} });
const bridge = new Bridge({ stdin: input, stdout: output, workDir: work, onChildReap: async () => {} });
const done = bridge.start(); input.write(JSON.stringify({ v: 1, type: 'start', id: `b_${++seq}`, payload: { ...start, remaining_ms: 150 } }) + '\n');
await assert.rejects(done, /deadline expired/); assert.equal(bridge.getState(), 'FAILED');
release(); assert(await bridge.flushOutput()); input.end();
assert.deepEqual(frames.map(f => f.type), ['model_request', 'collection_start', 'complete']);
});
test('EOF while complete is still queued fails before starting the terminal frame', async t => {
const work = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-queued-complete-'));
t.after(() => fs.rmSync(work, { recursive: true, force: true }));
fs.mkdirSync(path.join(work, 'artifacts')); fs.writeFileSync(path.join(work, 'artifacts', 'empty'), '');
const input = new PassThrough(); let seq = 0, release!: () => void; const frames: any[] = [];
let stored!: () => void; const acknowledged = new Promise<void>(resolve => stored = resolve);
const send = (type: string, payload: any, reply_to: string) => input.write(JSON.stringify({ v: 1, type, id: `b_${++seq}`, reply_to, payload }) + '\n');
const output = new Writable({ write(chunk, _encoding, callback) {
const f = JSON.parse(chunk.toString()); frames.push(f);
if (f.type === 'model_request') send('model_response', { result: final, error: null }, f.id);
if (f.type === 'collection_start') send('collection_ready', {}, f.id);
if (f.type === 'artifact_begin') send('artifact_ack', { transfer_id: f.payload.transfer_id, decision: 'accept', warning: null }, f.id);
if (f.type === 'artifact_end') {
release = () => callback();
send('artifact_ack', { transfer_id: f.payload.transfer_id, decision: 'stored', warning: null }, f.id); stored();
} else callback();
} });
const bridge = new Bridge({ stdin: input, stdout: output, workDir: work, onChildReap: async () => {} });
const done = bridge.start(); const failed = assert.rejects(done, /connection was lost/);
input.write(JSON.stringify({ v: 1, type: 'start', id: `b_${++seq}`, payload: start }) + '\n');
await acknowledged; await new Promise<void>(resolve => setImmediate(resolve));
input.end(); await failed; release(); assert(await bridge.flushOutput());
assert(!frames.some(f => f.type === 'complete')); assert.equal(frames.at(-1).type, 'error');
});

85
agent/tests/framing.test.ts Normal file
View File

@ -0,0 +1,85 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { PassThrough, Writable } from 'node:stream';
import { NDJsonFrameParser, StreamMessageWriter, sanitizeError } from '../framing.js';
import { envelope } from '../validation.js';
async function parse(chunks: Buffer[], max = 1024) {
const input = new PassThrough(); const parser = new NDJsonFrameParser(max);
input.pipe(parser);
const result = (async () => { const frames = []; for await (const frame of parser) frames.push(frame); return frames; })();
for (const c of chunks) input.write(c); input.end();
return result;
}
test('fragmented and coalesced UTF-8 frames, escaping', async () => {
const a = { v: 1, type: 'collection_ready', id: 'b_1', reply_to: 'a_1', payload: {} };
const b = { text: '雪 🐈 \\ " \n' };
const wire = Buffer.from(JSON.stringify(a) + '\n' + JSON.stringify(b) + '\n');
assert.deepEqual(await parse([...wire].map(x => Buffer.from([x]))), [a, b]);
assert.deepEqual(await parse([wire]), [a, b]);
});
test('invalid UTF-8, empty, malformed, oversized and truncated lines fail', async () => {
for (const bytes of [Buffer.from('\n'), Buffer.from('{\n'), Buffer.from('{"x":"\xff"}\n', 'latin1'), Buffer.alloc(65, 65), Buffer.from('{}')]) {
await assert.rejects(parse([bytes], 64));
}
assert.deepEqual(await parse([Buffer.from('{}\n')], 2), [{}]);
});
test('envelopes reject extra fields, wrong IDs and wrong versions', () => {
const a = { v: 1, type: 'collection_ready', id: 'b_1', reply_to: 'a_1', payload: {} };
envelope(a);
for (const b of [{ ...a, extra: true }, { ...a, v: 2 }, { ...a, id: 'a_1' }, { ...a, payload: { extra: true } }, { ...a, reply_to: 'invalid/path' }]) assert.throws(() => envelope(b));
});
test('serialized concurrent writes await slow callbacks and preserve order', async () => {
const lines: string[] = [];
const stream = new Writable({ highWaterMark: 1, write(chunk, _encoding, cb) { setTimeout(() => { lines.push(chunk.toString()); cb(); }, 5); } });
const writer = new StreamMessageWriter(stream);
await Promise.all(Array.from({ length: 8 }, (_, i) => writer.write({ v: 1, type: 'x', id: `a_${i}`, payload: {} })));
assert.deepEqual(lines.map(x => JSON.parse(x).id), Array.from({ length: 8 }, (_, i) => `a_${i}`));
});
test('terminal diagnostics never echo arbitrary exception content', () => {
assert(!JSON.stringify(sanitizeError('invalid_input', new Error('SECRET_PROMPT'))).includes('SECRET'));
assert.equal(sanitizeError('SECRET_CODE', 'SECRET_BODY').code, 'execution_failed');
});
test('fragment serializer preserves escaping and surrogate pairs across chunks', async () => {
const { jsonPieces, jsonBytes } = await import('../json.js');
const value = { text: 'x'.repeat(16383) + '🐈\u0001\n"\\', nested: [{ unicode: '雪' }, null, true, 3] };
const encoded = [...jsonPieces(value)].join('');
assert.equal(encoded, JSON.stringify(value));
assert.deepEqual(JSON.parse(encoded), value);
assert.equal(jsonBytes(value), Buffer.byteLength(encoded));
assert(jsonBytes(value, 32) > 32);
});
test('terminal write replaces backlog only after the active large NDJSON frame', async () => {
const chunks: string[] = []; let started!: () => void;
const first = new Promise<void>(resolve => started = resolve);
const stream = new Writable({ highWaterMark: 1, write(chunk, _encoding, cb) {
chunks.push(chunk.toString()); started(); setTimeout(cb, 8);
} });
const writer = new StreamMessageWriter(stream);
const large = { v: 1, type: 'collection_start', id: 'a_1', payload: { markdown: '雪'.repeat(100000), warnings: [] } };
const active = writer.write(large);
const queued = writer.write({ v: 1, type: 'artifact_begin', id: 'a_2', payload: {} });
const cancelled = assert.rejects(queued, /cancelled/);
await first;
const terminal = writer.writeTerminal({ v: 1, type: 'error', id: 'a_3', payload: sanitizeError('query_timeout') });
await Promise.all([active, cancelled, terminal, writer.close()]);
assert.deepEqual(chunks.join('').trimEnd().split('\n').map(line => JSON.parse(line)), [large, { v: 1, type: 'error', id: 'a_3', payload: sanitizeError('query_timeout') }]);
await assert.rejects(writer.write(large), /closed/);
});
test('a started terminal frame cannot be followed by another terminal message', async () => {
let release!: () => void; const chunks: string[] = [];
const stream = new Writable({ write(chunk, _encoding, cb) { chunks.push(chunk.toString()); release = () => cb(); } });
const writer = new StreamMessageWriter(stream);
const complete = writer.write({ v: 1, type: 'complete', id: 'a_1', payload: { accepted_transfer_count: 0 } });
await assert.rejects(writer.writeTerminal({ v: 1, type: 'error', id: 'a_2', payload: sanitizeError('query_timeout') }), /already started/);
release(); await complete; await writer.close();
assert.equal(chunks.join('').trimEnd().split('\n').length, 1);
});
test('broken output rejects active and queued frames and reports failed drain', async () => {
const stream = new Writable({ write(_chunk, _encoding, cb) { setTimeout(() => cb(new Error('DUMMY_PRIVATE_ERROR')), 5); } });
const writer = new StreamMessageWriter(stream);
await Promise.all([1, 2].map(i => assert.rejects(writer.write({ v: 1, type: 'x', id: `a_${i}`, payload: {} }), /Output stream failed/)));
await assert.rejects(writer.close(), /Output stream failed/);
});

19
agent/tests/local-tools.test.ts Normal file
View File

@ -0,0 +1,19 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { localTools } from '../local-tools.js';
test('native shell bounds output and reports timeout as a failed tool', async t => {
const work = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-local-')); t.after(() => fs.rmSync(work, { recursive: true, force: true }));
const bash = localTools(work).find(tool => tool.name === 'bash')!;
await assert.rejects(bash.execute('sdk_output', { command: "python3 -c 'import sys; sys.stdout.write(\"x\"*2000000)'" }), /capped at 1 MiB/);
await assert.rejects(bash.execute('sdk_timeout', { command: 'sleep 10', timeout: 0.05 }), /timed out/);
});
test('native reads reject large and nonregular files without blocking', async t => {
const work = fs.mkdtempSync(path.join(os.tmpdir(), 'pi-read-')); t.after(() => fs.rmSync(work, { recursive: true, force: true }));
const read = localTools(work).find(tool => tool.name === 'read')!;
fs.writeFileSync(path.join(work, 'large'), ''); fs.truncateSync(path.join(work, 'large'), 17 * 1024 * 1024);
await assert.rejects(read.execute('sdk_large', { path: path.join(work, 'large') }), /at most 16 MiB/);
await assert.rejects(read.execute('sdk_special', { path: '/dev/zero' }), /regular file/);
});

30
agent/tests/peer.test.ts Normal file
View File

@ -0,0 +1,30 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { runPeer } from '../dev/fake-backend.js';
for (const mode of ['happy', 'no-artifacts', 'empty-file', 'skip', 'delayed', 'background', 'spaces', 'confluence-error']) {
test(`real SDK and native tools over public wire: ${mode}`, async () => { assert((await runPeer({ mode })).complete); });
}
for (const mode of ['upstream-failure', 'bad-reply', 'duplicate', 'eof-model', 'context-limit', 'output-limit', 'zero-text']) {
test(`scripted peer failure: ${mode}`, async () => {
const result = await runPeer({ mode }); assert(!result.complete);
if (mode !== 'eof-model') assert(result.failure, result.stderr);
assert(!JSON.stringify(result.failure).includes('DUMMY_PRIVATE'));
});
}
test('host supervisor leaves sibling processes alive', async t => {
const { spawn } = await import('node:child_process');
const sibling = spawn('sleep', ['10']);
t.after(() => sibling.kill('SIGTERM'));
assert((await runPeer()).complete);
assert.equal(sibling.exitCode, null);
assert.equal(sibling.signalCode, null);
});
test('failure entrypoint flushes an active large line and error through a slow peer', async () => {
const result = await runPeer({ mode: 'failure-during-collection' });
assert(!result.complete); assert.equal(result.failure.code, 'invalid_input');
const outgoing = result.transcript.filter(f => f.id.startsWith('a_'));
assert.deepEqual(outgoing.slice(-2).map(f => f.type), ['collection_start', 'error']);
assert(outgoing.at(-2).payload.markdown.length > 1024 * 1024);
});

16
agent/tests/sdk.test.ts Normal file
View File

@ -0,0 +1,16 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { contextToModelRequest, modelResponseToAssistantMessage, createModelProvider } from '../model-provider.js';
const descriptor = { id: 'scripted', context_window_tokens: 1000, max_output_tokens: 100 };
test('usage, stop reasons and opaque continuation survive SDK translation', () => {
const { model } = createModelProvider(descriptor, async () => { throw Error(); });
for (const reason of ['stop', 'tool_calls', 'length'] as const) {
const response = { content: [{ type: 'text' as const, text: 'hi' }, { type: 'tool_call' as const, id: 'sdk_1', name: 'read', arguments: { path: '/work/x' } }], stop_reason: reason, usage: { input_tokens: 7, output_tokens: 3 }, provider_state: '' };
const message = modelResponseToAssistantMessage(response, model);
assert.equal(message.usage.totalTokens, 10);
assert.equal(message.stopReason, reason === 'tool_calls' ? 'toolUse' : reason);
const request = contextToModelRequest({ systemPrompt: 'TRUSTED', messages: [message] });
assert.deepEqual(request.messages[0], { role: 'assistant', content: response.content, provider_state: '' });
assert(!JSON.stringify(request).includes('TRUSTED'));
}
});

23
agent/tests/warnings.test.ts Normal file
View File

@ -0,0 +1,23 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { WarningCollector } from '../warnings.js';
import { warning } from '../validation.js';
test('repeats preserve bounded Unicode messages and snapshots do not alias', () => {
const collector = new WarningCollector(); const value = { code: 'repeat', message: '雪'.repeat(341), name: 'file' };
collector.add(value); const before = collector.snapshot();
for (let i = 0; i < 150; i++) collector.add(value);
const entries = collector.snapshot(); assert.equal(entries.length, 1); warning(entries[0]);
assert(Buffer.byteLength(entries[0].message) <= 1024); assert(!entries[0].message.includes('<27>')); assert.match(entries[0].message, /Repeated 151 times/);
assert.equal(before[0].message, value.message);
});
test('overflow includes repetitions of the replaced last warning', () => {
const collector = new WarningCollector();
for (let i = 0; i < 100; i++) collector.add({ code: `warning_${i}`, message: 'Warning.' });
collector.add({ code: 'warning_99', message: 'Warning.' });
collector.add({ code: 'overflow', message: 'Warning.' });
collector.add({ code: 'warning_99', message: 'Warning.' });
const entries = collector.snapshot(); assert.equal(entries.length, 100);
assert.match(entries.at(-1)!.message, /^4 additional/);
entries.forEach(warning);
});

25
agent/tsconfig.json Normal file
View File

@ -0,0 +1,25 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": ["ES2022"],
"outDir": "./dist",
"rootDir": "./",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"declaration": true,
"sourceMap": true
},
"include": [
"*.ts",
"dev/**/*.ts",
"tests/**/*.ts"
],
"exclude": [
"node_modules",
"dist"
]
}

252
agent/types.ts Normal file
View File

@ -0,0 +1,252 @@
/**
* Protocol types and limits defined in CONTRACTS.md Revision 1
*/
export const PROTOCOL_VERSION = 1;
// Size limits in bytes
export const KIB = 1024;
export const MIB = 1024 * 1024;
export const LIMITS = {
USER_PROMPT_MAX_BYTES: 16 * MIB,
FINAL_MARKDOWN_MAX_BYTES: 128 * MIB,
TOOL_REQUEST_PAYLOAD_MAX_BYTES: 16 * MIB,
TOOL_RESULT_PAYLOAD_MAX_BYTES: 128 * MIB,
MODEL_REQUEST_PAYLOAD_MAX_BYTES: 128 * MIB,
MODEL_RESULT_PAYLOAD_MAX_BYTES: 128 * MIB,
ENVELOPE_ALLOWANCE_BYTES: 64 * KIB,
START_FRAME_MAX_BYTES: 6 * (16 * MIB) + 64 * KIB,
COLLECTION_START_FRAME_MAX_BYTES: 6 * (128 * MIB) + 64 * KIB,
ORDINARY_FRAME_MAX_BYTES: 128 * MIB + 64 * KIB,
ARTIFACT_CHUNK_MAX_DECODED_BYTES: 64 * KIB,
ARTIFACT_MAX_FILES: 20,
ARTIFACT_MAX_FILE_BYTES: 10 * MIB,
ARTIFACT_MAX_TOTAL_BYTES: 50 * MIB,
MAX_DEADLINE_MS: 180_000,
MAX_REMOTE_CONCURRENCY: 4,
MAX_WARNINGS: 100,
MAX_ERROR_MESSAGE_BYTES: 1024,
MAX_PATH_BYTES: 1024
} as const;
export interface ErrorShape {
code: string;
message: string;
}
export interface Warning {
code: string;
message: string;
tool_call_id?: string;
name?: string;
}
export interface Page {
page_id: string;
title: string;
space: string;
url: string;
}
export interface Pagination {
offset: number;
limit: number;
has_more: boolean;
}
export interface ModelDescriptor {
id: string;
context_window_tokens: number;
max_output_tokens: number;
}
// Model Protocol
export type TextContentBlock = {
type: "text";
text: string;
};
export type ToolCallContentBlock = {
type: "tool_call";
id: string;
name: string;
arguments: Record<string, any>;
};
export type ContentBlock = TextContentBlock | ToolCallContentBlock;
export interface UserMessageContract {
role: "user";
content: Array<{ type: "text"; text: string }>;
}
export interface AssistantMessageContract {
role: "assistant";
content: ContentBlock[];
provider_state?: string;
}
export interface ToolMessageContract {
role: "tool";
tool_call_id: string;
name: string;
content: string;
is_error: boolean;
}
export type MessageContract =
| UserMessageContract
| AssistantMessageContract
| ToolMessageContract;
export interface ToolDeclaration {
name: string;
description: string;
input_schema: Record<string, any>;
}
export interface ModelRequest {
messages: MessageContract[];
tools: ToolDeclaration[];
}
export interface ModelResponse {
content: ContentBlock[];
stop_reason: "stop" | "tool_calls" | "length";
usage: {
input_tokens: number;
output_tokens: number;
};
provider_state?: string;
}
// Bridge Protocol Frames
export interface BridgeBaseMessage {
v: number;
type: string;
id: string;
reply_to?: string;
payload: any;
}
export interface StartMessage extends BridgeBaseMessage {
type: "start";
payload: {
prompt: string;
system_instruction: string;
remaining_ms: number;
model: ModelDescriptor;
};
}
export interface ToolRequestMessage extends BridgeBaseMessage {
type: "tool_request";
payload: {
tool: string;
parameters: Record<string, any>;
};
}
export interface ToolResponseMessage extends BridgeBaseMessage {
type: "tool_response";
reply_to: string;
payload: {
result: any;
error: ErrorShape | null;
};
}
export interface ModelRequestMessage extends BridgeBaseMessage {
type: "model_request";
payload: ModelRequest;
}
export interface ModelResponseMessage extends BridgeBaseMessage {
type: "model_response";
reply_to: string;
payload: {
result: ModelResponse | null;
error: ErrorShape | null;
};
}
export interface CollectionStartMessage extends BridgeBaseMessage {
type: "collection_start";
payload: {
markdown: string;
warnings: Warning[];
};
}
export interface CollectionReadyMessage extends BridgeBaseMessage {
type: "collection_ready";
reply_to: string;
payload: Record<string, never>;
}
export interface ArtifactBeginMessage extends BridgeBaseMessage {
type: "artifact_begin";
payload: {
transfer_id: string;
name: string;
size_bytes: number;
};
}
export interface ArtifactAckMessage extends BridgeBaseMessage {
type: "artifact_ack";
reply_to: string;
payload: {
transfer_id: string;
decision: "accept" | "skip" | "stored";
warning: Warning | null;
};
}
export interface ArtifactChunkMessage extends BridgeBaseMessage {
type: "artifact_chunk";
payload: {
transfer_id: string;
index: number;
data_base64: string;
};
}
export interface ArtifactEndMessage extends BridgeBaseMessage {
type: "artifact_end";
payload: {
transfer_id: string;
size_bytes: number;
chunks: number;
};
}
export interface CompleteMessage extends BridgeBaseMessage {
type: "complete";
payload: {
accepted_transfer_count: number;
};
}
export interface ErrorMessage extends BridgeBaseMessage {
type: "error";
payload: ErrorShape;
}
export type BridgeInboundMessage =
| StartMessage
| ToolResponseMessage
| ModelResponseMessage
| CollectionReadyMessage
| ArtifactAckMessage;
export type BridgeOutboundMessage =
| ToolRequestMessage
| ModelRequestMessage
| CollectionStartMessage
| ArtifactBeginMessage
| ArtifactChunkMessage
| ArtifactEndMessage
| CompleteMessage
| ErrorMessage;

101
agent/validation.ts Normal file
View File

@ -0,0 +1,101 @@
import { jsonBytes } from './json.js';
import { LIMITS } from './types.js';
import { isValidId } from './framing.js';
export function object(x: unknown): x is Record<string, any> {
return !!x && typeof x === 'object' && !Array.isArray(x);
}
export function keys(x: unknown, required: string[], optional: string[] = []): asserts x is Record<string, any> {
if (!object(x) || required.some(k => !Object.hasOwn(x, k)) || Object.keys(x).some(k => !required.includes(k) && !optional.includes(k))) throw new Error('Invalid fields');
}
export function integer(x: unknown, min = 0, max = Number.MAX_SAFE_INTEGER): void {
if (!Number.isSafeInteger(x) || (x as number) < min || (x as number) > max) throw new Error('Invalid integer');
}
export function text(x: unknown, max: number, nonempty = false): void {
if (typeof x !== 'string' || Buffer.byteLength(x) > max || (nonempty && !x.trim())) throw new Error('Invalid text');
}
export function warning(x: unknown): void {
keys(x, ['code', 'message'], ['tool_call_id', 'name']);
text(x.code, 128, true); text(x.message, 1024);
if (x.tool_call_id !== undefined && !isValidId(x.tool_call_id)) throw new Error('Invalid warning ID');
if (x.name !== undefined) text(x.name, 1024);
}
export function envelope(x: any): void {
keys(x, ['v', 'type', 'id', 'payload'], ['reply_to']);
if (x.v !== 1 || !isValidId(x.id) || !x.id.startsWith('b_')) throw new Error('Invalid envelope');
if (x.type === 'start') {
if (Object.hasOwn(x, 'reply_to')) throw new Error('Unexpected correlation');
keys(x.payload, ['prompt', 'system_instruction', 'remaining_ms', 'model']);
text(x.payload.prompt, LIMITS.USER_PROMPT_MAX_BYTES, true);
text(x.payload.system_instruction, LIMITS.ENVELOPE_ALLOWANCE_BYTES, true);
integer(x.payload.remaining_ms, 1, LIMITS.MAX_DEADLINE_MS);
const m = x.payload.model;
keys(m, ['id', 'context_window_tokens', 'max_output_tokens']);
text(m.id, 1024, true); integer(m.context_window_tokens, 1); integer(m.max_output_tokens, 1, m.context_window_tokens);
const metadata = { ...x, payload: { ...x.payload, prompt: '' } };
if (jsonBytes(metadata, LIMITS.ENVELOPE_ALLOWANCE_BYTES) > LIMITS.ENVELOPE_ALLOWANCE_BYTES) throw new Error('Start metadata too large');
} else {
if (!isValidId(x.reply_to) || !x.reply_to.startsWith('a_')) throw new Error('Invalid correlation');
if (x.type === 'collection_ready') keys(x.payload, []);
else if (x.type === 'artifact_ack') {
keys(x.payload, ['transfer_id', 'decision', 'warning']);
if (!isValidId(x.payload.transfer_id) || !['accept', 'skip', 'stored'].includes(x.payload.decision)) throw new Error('Invalid acknowledgement');
if (x.payload.warning !== null) warning(x.payload.warning);
if (x.payload.decision === 'stored' && x.payload.warning !== null) throw new Error('Unexpected stored warning');
} else if (x.type === 'tool_response' || x.type === 'model_response') {
keys(x.payload, ['result', 'error']);
if (x.payload.error !== null) {
keys(x.payload.error, ['code', 'message']); text(x.payload.error.code, 128, true); text(x.payload.error.message, 1024);
if (x.payload.result !== null) throw new Error('Invalid error result');
} else {
if (!object(x.payload.result)) throw new Error('Invalid success result');
if (x.type === 'model_response') modelResponse(x.payload.result);
}
if (jsonBytes(x.payload, LIMITS.MODEL_RESULT_PAYLOAD_MAX_BYTES) > LIMITS.MODEL_RESULT_PAYLOAD_MAX_BYTES) throw new Error('Payload too large');
} else throw new Error('Unknown frame');
}
}
export function modelResponse(x: any): void {
keys(x, ['content', 'stop_reason', 'usage'], ['provider_state']);
if (!Array.isArray(x.content) || !['stop', 'tool_calls', 'length'].includes(x.stop_reason)) throw new Error('Invalid model result');
const ids = new Set<string>();
for (const c of x.content) {
if (c.type === 'text') { keys(c, ['type', 'text']); text(c.text, LIMITS.FINAL_MARKDOWN_MAX_BYTES); }
else if (c.type === 'tool_call') {
keys(c, ['type', 'id', 'name', 'arguments']);
if (!isValidId(c.id) || ids.has(c.id) || !object(c.arguments)) throw new Error('Invalid model call');
ids.add(c.id); text(c.name, 128, true);
if (jsonBytes(c.arguments, LIMITS.TOOL_REQUEST_PAYLOAD_MAX_BYTES) > LIMITS.TOOL_REQUEST_PAYLOAD_MAX_BYTES) throw new Error('Tool arguments too large');
} else throw new Error('Invalid content');
}
if ((ids.size > 0) !== (x.stop_reason === 'tool_calls') && x.stop_reason !== 'length') throw new Error('Invalid stop reason');
keys(x.usage, ['input_tokens', 'output_tokens']); integer(x.usage.input_tokens); integer(x.usage.output_tokens);
if (x.provider_state !== undefined) text(x.provider_state, 64 * 1024);
}
export function confluenceResult(x: any, tool: string): void {
const page = (p: any, extra: string[]) => {
keys(p, ['page_id', 'title', 'space', 'url', ...extra]);
text(p.page_id, 128, true);
if (!/^[0-9]+$/.test(p.page_id)) throw new Error('Invalid page ID');
for (const field of ['title', 'space', 'url']) text(p[field], LIMITS.TOOL_RESULT_PAYLOAD_MAX_BYTES);
};
const pagination = (p: any, count: number) => {
keys(p, ['offset', 'limit', 'has_more']);
integer(p.offset, 0, 10000); integer(p.limit, 1, 50);
if (typeof p.has_more !== 'boolean' || count > p.limit || (!count && p.has_more)) throw new Error('Invalid pagination');
};
if (tool === 'confluence_view') {
page(x, ['markdown', 'truncated']); text(x.markdown, LIMITS.TOOL_RESULT_PAYLOAD_MAX_BYTES);
if (typeof x.truncated !== 'boolean') throw new Error('Invalid truncation');
} else if (tool === 'confluence_search') {
keys(x, ['pages', 'pagination']);
if (!Array.isArray(x.pages) || x.pages.length > 50) throw new Error('Invalid pages');
for (const p of x.pages) { page(p, ['snippet']); text(p.snippet, LIMITS.TOOL_RESULT_PAYLOAD_MAX_BYTES); }
pagination(x.pagination, x.pages.length);
} else if (tool === 'confluence_list_spaces') {
keys(x, ['spaces', 'pagination']);
if (!Array.isArray(x.spaces) || x.spaces.length > 50) throw new Error('Invalid spaces');
for (const s of x.spaces) { keys(s, ['key', 'name']); text(s.key, LIMITS.TOOL_RESULT_PAYLOAD_MAX_BYTES); text(s.name, LIMITS.TOOL_RESULT_PAYLOAD_MAX_BYTES); }
pagination(x.pagination, x.spaces.length);
} else throw new Error('Invalid Confluence operation');
}

40
agent/warnings.ts Normal file
View File

@ -0,0 +1,40 @@
import { LIMITS, type Warning } from './types.js';
function boundedMessage(message: string, suffix: string): string {
const bytes = Buffer.from(message);
let end = Math.min(bytes.length, LIMITS.MAX_ERROR_MESSAGE_BYTES - Buffer.byteLength(suffix));
while (end < bytes.length && (bytes[end] & 0xc0) === 0x80) end--;
return bytes.subarray(0, end).toString('utf8') + suffix;
}
/** Keep repeated warnings and overflow visible without extending the wire shape. */
export class WarningCollector {
private entries: Warning[] = [];
private groups = new Map<string, { warning: Warning; message: string; count: number }>();
private overflow = 0;
add(warning: Warning): void {
const key = JSON.stringify([warning.code, warning.message, warning.tool_call_id, warning.name]);
const existing = this.groups.get(key);
if (existing) {
existing.count++;
existing.warning.message = boundedMessage(existing.message, ` (Repeated ${existing.count} times.)`);
return;
}
if (!this.overflow && this.entries.length < LIMITS.MAX_WARNINGS) {
const copy = { ...warning };
this.entries.push(copy);
this.groups.set(key, { warning: copy, message: copy.message, count: 1 });
return;
}
if (!this.overflow) {
const replaced = this.entries.pop()!;
const removed = [...this.groups.entries()].find(([, group]) => group.warning === replaced)!;
this.overflow = removed[1].count;
this.groups.delete(removed[0]);
this.entries.push({ code: 'warnings_aggregated', message: '' });
}
this.overflow++;
this.entries.at(-1)!.message = `${this.overflow} additional warnings were aggregated after the warning limit.`;
}
snapshot(): Warning[] { return this.entries.map(w => ({ ...w })); }
}