feat: add turn streaming and inspector updates

This commit is contained in:
Nathan Flurry 2026-01-27 06:18:43 -08:00
parent bf58891edf
commit 34d4f3693e
49 changed files with 4629 additions and 1146 deletions

View file

@ -24,6 +24,7 @@ Research on how different agents operate (CLI flags, streaming formats, HITL pat
Universal schema guidance:
- The universal schema should cover the full feature set of all agents.
- Conversions must be best-effort overlap without being lossy; preserve raw payloads when needed.
- **The mock agent acts as the reference implementation** for correct event behavior. Real agents should use synthetic events to match the mock agent's event patterns (e.g., emitting both daemon synthetic and agent native `session.started` events, proper `item.started``item.delta``item.completed` sequences).
## Spec Tracking
@ -54,6 +55,7 @@ Universal schema guidance:
- `sandbox-agent sessions list``GET /v1/sessions`
- `sandbox-agent sessions create``POST /v1/sessions/{sessionId}`
- `sandbox-agent sessions send-message``POST /v1/sessions/{sessionId}/messages`
- `sandbox-agent sessions send-message-stream``POST /v1/sessions/{sessionId}/messages/stream`
- `sandbox-agent sessions events` / `get-messages``GET /v1/sessions/{sessionId}/events`
- `sandbox-agent sessions events-sse``GET /v1/sessions/{sessionId}/events/sse`
- `sandbox-agent sessions reply-question``POST /v1/sessions/{sessionId}/questions/{questionId}/reply`

View file

@ -150,6 +150,7 @@ Create a session and send a message:
```bash
sandbox-agent sessions create my-session --agent codex --endpoint http://127.0.0.1:2468 --token "$SANDBOX_TOKEN"
sandbox-agent sessions send-message my-session --message "Hello" --endpoint http://127.0.0.1:2468 --token "$SANDBOX_TOKEN"
sandbox-agent sessions send-message-stream my-session --message "Hello" --endpoint http://127.0.0.1:2468 --token "$SANDBOX_TOKEN"
```
Docs: https://rivet.dev/docs/cli

View file

@ -105,6 +105,7 @@ Each session tracks:
POST /v1/sessions/{sessionId} Create session, auto-install agent
POST /v1/sessions/{id}/messages Spawn agent subprocess, stream output
POST /v1/sessions/{id}/messages/stream Post and stream a single turn
GET /v1/sessions/{id}/events Poll for new events (offset-based)
GET /v1/sessions/{id}/events/sse Subscribe to SSE stream
@ -133,16 +134,30 @@ When a message is sent:
## Agent Execution
Each agent has a different execution model and communication pattern.
Each agent has a different execution model and communication pattern. There are two main architectural patterns:
### Architecture Patterns
**Subprocess Model (Claude, Amp):**
- New process spawned per message/turn
- Process terminates after turn completes
- Multi-turn via CLI resume flags (`--resume`, `--continue`)
- Simple but has process spawn overhead
**Client/Server Model (OpenCode, Codex):**
- Single long-running server process
- Multiple sessions/threads multiplexed via RPC
- Multi-turn via server-side thread persistence
- More efficient for repeated interactions
### Overview
| Agent | Execution Model | Binary Source | Session Resume |
|-------|-----------------|---------------|----------------|
| Claude Code | CLI subprocess | GCS (Anthropic) | Yes (`--resume`) |
| Codex | App Server subprocess (JSON-RPC) | GitHub releases | No |
| OpenCode | HTTP server + SSE | GitHub releases | Yes (server-side) |
| Amp | CLI subprocess | GCS (Amp) | Yes (`--continue`) |
| Agent | Architecture | Binary Source | Multi-Turn Method |
|-------|--------------|---------------|-------------------|
| Claude Code | Subprocess (per-turn) | GCS (Anthropic) | `--resume` flag |
| Codex | **Shared Server (JSON-RPC)** | GitHub releases | **Thread persistence** |
| OpenCode | HTTP Server (SSE) | GitHub releases | Server-side sessions |
| Amp | Subprocess (per-turn) | GCS (Amp) | `--continue` flag |
### Claude Code
@ -161,15 +176,25 @@ claude --print --output-format stream-json --verbose \
### Codex
Spawned as a subprocess using the App Server JSON-RPC protocol:
Uses a **shared app-server process** that handles multiple sessions via JSON-RPC over stdio:
```bash
codex app-server
```
- JSON-RPC over stdio (JSONL)
- Uses `initialize`, `thread/start`, and `turn/start` requests
- Approval requests arrive as server JSON-RPC requests
**Daemon flow:**
1. First Codex session triggers `codex app-server` spawn
2. Performs `initialize` / `initialized` handshake
3. Each session creation sends `thread/start` → receives `thread_id`
4. Messages sent via `turn/start` with `thread_id`
5. Notifications routed back to session by `thread_id`
**Key characteristics:**
- Single process handles all Codex sessions
- JSON-RPC over stdio (JSONL format)
- Thread IDs map to daemon session IDs
- Approval requests arrive as server-to-client JSON-RPC requests
- Process lifetime matches daemon lifetime (not per-turn)
### OpenCode
@ -208,12 +233,21 @@ amp [--execute|--print] [--output-format stream-json] \
### Communication Patterns
**Subprocess agents (Claude, Codex, Amp):**
**Per-turn subprocess agents (Claude, Amp):**
1. Agent CLI spawned with appropriate flags
2. Stdout/stderr read line-by-line
3. Each line parsed as JSON
4. Events converted via `parse_agent_line()` → agent-specific converter
5. Universal events recorded and broadcast to SSE subscribers
6. Process terminated on turn completion
**Shared stdio server agent (Codex):**
1. Single `codex app-server` process started on first session
2. `initialize`/`initialized` handshake performed once
3. New sessions send `thread/start`, receive `thread_id`
4. Messages sent via `turn/start` with `thread_id`
5. Notifications read from stdout, routed by `thread_id`
6. Process persists across sessions and turns
**HTTP server agent (OpenCode):**
1. Server started on available port (if not running)

View file

@ -131,13 +131,15 @@ timestamps, not ordering.
## Optional raw payloads
If you need provider-level debugging, pass `include_raw=true` when streaming or polling events to
receive the `raw` payload for each event.
If you need provider-level debugging, pass `include_raw=true` when streaming or polling events
(including one-turn streams) to receive the `raw` payload for each event.
## SSE vs polling
## SSE vs polling vs turn streaming
- SSE gives low-latency updates and simplifies streaming UIs.
- Polling is simpler to debug and works in any environment.
- Turn streaming (`POST /v1/sessions/{session_id}/messages/stream`) is a one-shot stream tied to a
single prompt. The stream closes automatically once the turn completes.
Both yield the same event payloads.

View file

@ -67,6 +67,16 @@ sandbox-agent sessions send-message my-session \
```
</details>
<details>
<summary><strong>sessions send-message-stream</strong></summary>
```bash
sandbox-agent sessions send-message-stream my-session \
--message "Summarize the repository" \
--endpoint http://127.0.0.1:2468
```
</details>
<details>
<summary><strong>sessions events</strong></summary>

View file

@ -408,6 +408,60 @@
}
}
},
"/v1/sessions/{session_id}/messages/stream": {
"post": {
"tags": [
"sessions"
],
"operationId": "post_message_stream",
"parameters": [
{
"name": "session_id",
"in": "path",
"description": "Session id",
"required": true,
"schema": {
"type": "string"
}
},
{
"name": "include_raw",
"in": "query",
"description": "Include raw provider payloads",
"required": false,
"schema": {
"type": "boolean",
"nullable": true
}
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MessageRequest"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "SSE event stream"
},
"404": {
"description": "",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
}
}
}
},
"/v1/sessions/{session_id}/permissions/{permission_id}/reply": {
"post": {
"tags": [
@ -1431,6 +1485,15 @@
"daemon"
]
},
"TurnStreamQuery": {
"type": "object",
"properties": {
"includeRaw": {
"type": "boolean",
"nullable": true
}
}
},
"UniversalEvent": {
"type": "object",
"required": [

View file

@ -70,6 +70,15 @@ curl "http://127.0.0.1:2468/v1/sessions/my-session/events/sse?offset=0" \
-H "Authorization: Bearer $SANDBOX_TOKEN"
```
For a single-turn stream (post a message and get one streamed response):
```bash
curl -N -X POST "http://127.0.0.1:2468/v1/sessions/my-session/messages/stream" \
-H "Authorization: Bearer $SANDBOX_TOKEN" \
-H "content-type: application/json" \
-d '{"message":"Hello"}'
```
## 5. CLI shortcuts
The CLI mirrors the HTTP API:
@ -78,4 +87,6 @@ The CLI mirrors the HTTP API:
sandbox-agent sessions create my-session --agent claude --endpoint http://127.0.0.1:2468 --token "$SANDBOX_TOKEN"
sandbox-agent sessions send-message my-session --message "Hello" --endpoint http://127.0.0.1:2468 --token "$SANDBOX_TOKEN"
sandbox-agent sessions send-message-stream my-session --message "Hello" --endpoint http://127.0.0.1:2468 --token "$SANDBOX_TOKEN"
```

View file

@ -86,10 +86,21 @@ for await (const event of client.streamEvents("demo-session", {
The SDK parses `text/event-stream` into `UniversalEvent` objects. If you want full control, use
`getEventsSse()` and parse the stream yourself.
## Stream a single turn
```ts
for await (const event of client.streamTurn("demo-session", { message: "Hello" })) {
console.log(event.type, event.data);
}
```
This method posts the message and streams only the next turn. For manual control, call
`postMessageStream()` and parse the SSE response yourself.
## Optional raw payloads
Set `includeRaw: true` on `getEvents` or `streamEvents` to include the raw provider payload in
`event.raw`. This is useful for debugging and conversion analysis.
Set `includeRaw: true` on `getEvents`, `streamEvents`, or `streamTurn` to include the raw provider
payload in `event.raw`. This is useful for debugging and conversion analysis.
## Error handling

View file

@ -3,7 +3,7 @@ title: "Telemetry"
description: "Anonymous telemetry collected by sandbox-agent."
---
sandbox-agent sends a small, anonymous telemetry payload on startup to help us understand usage and improve reliability.
sandbox-agent sends a small, anonymous telemetry payload on startup and then every 5 minutes to help us understand usage and improve reliability.
## What gets sent
@ -12,6 +12,7 @@ sandbox-agent sends a small, anonymous telemetry payload on startup to help us u
- Detected sandbox provider (for example: Docker, E2B, Vercel Sandboxes).
Each sandbox gets a random anonymous ID stored on disk so usage can be counted without identifying users.
The last successful send time is also stored on disk, and heartbeats are rate-limited to at most one every 5 minutes.
## Opting out

View file

@ -383,7 +383,7 @@
flex-direction: column;
border-right: 1px solid var(--border);
background: var(--surface-2);
overflow: hidden;
overflow: visible;
}
.sidebar-header {
@ -394,12 +394,15 @@
align-items: center;
justify-content: space-between;
flex-shrink: 0;
overflow: visible;
}
.sidebar-header-actions {
display: flex;
align-items: center;
gap: 8px;
position: relative;
overflow: visible;
}
.sidebar-icon-btn {
@ -449,6 +452,53 @@
background: var(--accent-hover);
}
.sidebar-add-menu-wrapper {
position: relative;
}
.sidebar-add-menu {
position: absolute;
top: 30px;
right: 0;
min-width: 140px;
background: var(--surface);
border: 1px solid var(--border-2);
border-radius: 8px;
box-shadow: 0 8px 24px rgba(0, 0, 0, 0.35);
padding: 6px;
display: flex;
flex-direction: column;
gap: 4px;
z-index: 60;
}
.sidebar-add-option {
background: transparent;
border: 1px solid transparent;
color: var(--text);
text-align: left;
padding: 6px 8px;
border-radius: 6px;
font-size: 12px;
cursor: pointer;
transition: all var(--transition);
}
.sidebar-add-option:hover {
background: var(--accent);
color: #fff;
}
.sidebar-add-status {
padding: 6px 8px;
font-size: 11px;
color: var(--muted);
}
.sidebar-add-status.error {
color: var(--danger);
}
.session-list {
flex: 1;
overflow-y: auto;
@ -520,6 +570,10 @@
font-size: 11px;
}
.sidebar-empty.error {
color: var(--danger);
}
/* Chat Panel */
.chat-panel {
display: flex;
@ -560,6 +614,21 @@
color: var(--text-secondary);
}
.session-agent-display {
font-size: 11px;
font-weight: 600;
color: var(--accent);
background: color-mix(in srgb, var(--accent) 18%, transparent);
padding: 2px 6px;
border-radius: 999px;
}
.panel-header-right {
display: flex;
align-items: center;
gap: 8px;
}
.messages-container {
flex: 1;
overflow-y: auto;
@ -947,6 +1016,13 @@
width: 50px;
}
.setup-select-small:disabled,
.setup-select:disabled,
.setup-input:disabled {
opacity: 0.55;
cursor: not-allowed;
}
.setup-stream-btn {
display: flex;
align-items: center;
@ -973,6 +1049,16 @@
color: var(--accent);
}
.setup-stream-btn:disabled {
cursor: default;
opacity: 0.6;
}
.setup-stream-btn:disabled:hover {
border-color: var(--border-2);
color: var(--muted);
}
.setup-stream-btn.active {
background: var(--accent);
border-color: var(--accent);

View file

@ -62,6 +62,12 @@ export default function App() {
const [agents, setAgents] = useState<AgentInfo[]>([]);
const [modesByAgent, setModesByAgent] = useState<Record<string, AgentModeInfo[]>>({});
const [sessions, setSessions] = useState<SessionInfo[]>([]);
const [agentsLoading, setAgentsLoading] = useState(false);
const [agentsError, setAgentsError] = useState<string | null>(null);
const [sessionsLoading, setSessionsLoading] = useState(false);
const [sessionsError, setSessionsError] = useState<string | null>(null);
const [modesLoadingByAgent, setModesLoadingByAgent] = useState<Record<string, boolean>>({});
const [modesErrorByAgent, setModesErrorByAgent] = useState<Record<string, string | null>>({});
const [agentId, setAgentId] = useState("claude");
const [agentMode, setAgentMode] = useState("");
@ -75,10 +81,12 @@ export default function App() {
const [events, setEvents] = useState<UniversalEvent[]>([]);
const [offset, setOffset] = useState(0);
const offsetRef = useRef(0);
const [eventsLoading, setEventsLoading] = useState(false);
const [polling, setPolling] = useState(false);
const pollTimerRef = useRef<number | null>(null);
const [streamMode, setStreamMode] = useState<"poll" | "sse">("sse");
const [turnStreaming, setTurnStreaming] = useState(false);
const [streamMode, setStreamMode] = useState<"poll" | "sse" | "turn">("sse");
const [eventError, setEventError] = useState<string | null>(null);
const [questionSelections, setQuestionSelections] = useState<Record<string, string[][]>>({});
@ -95,6 +103,7 @@ export default function App() {
const clientRef = useRef<SandboxAgent | null>(null);
const sseAbortRef = useRef<AbortController | null>(null);
const turnAbortRef = useRef<AbortController | null>(null);
const logRequest = useCallback((entry: RequestLog) => {
setRequestLog((prev) => {
@ -200,9 +209,18 @@ export default function App() {
setEventError(null);
stopPolling();
stopSse();
stopTurnStream();
setAgents([]);
setSessions([]);
setAgentsLoading(false);
setSessionsLoading(false);
setAgentsError(null);
setSessionsError(null);
};
const refreshAgents = async () => {
setAgentsLoading(true);
setAgentsError(null);
try {
const data = await getClient().listAgents();
const agentList = data.agents ?? [];
@ -213,17 +231,23 @@ export default function App() {
}
}
} catch (error) {
setConnectError(getErrorMessage(error, "Unable to refresh agents"));
setAgentsError(getErrorMessage(error, "Unable to refresh agents"));
} finally {
setAgentsLoading(false);
}
};
const fetchSessions = async () => {
setSessionsLoading(true);
setSessionsError(null);
try {
const data = await getClient().listSessions();
const sessionList = data.sessions ?? [];
setSessions(sessionList);
} catch {
// Silently fail - sessions list is supplementary
setSessionsError("Unable to load sessions.");
} finally {
setSessionsLoading(false);
}
};
@ -237,22 +261,32 @@ export default function App() {
};
const loadModes = async (targetId: string) => {
setModesLoadingByAgent((prev) => ({ ...prev, [targetId]: true }));
setModesErrorByAgent((prev) => ({ ...prev, [targetId]: null }));
try {
const data = await getClient().getAgentModes(targetId);
const modes = data.modes ?? [];
setModesByAgent((prev) => ({ ...prev, [targetId]: modes }));
} catch {
// Silently fail - modes are optional
setModesErrorByAgent((prev) => ({ ...prev, [targetId]: "Unable to load modes." }));
} finally {
setModesLoadingByAgent((prev) => ({ ...prev, [targetId]: false }));
}
};
const sendMessage = async () => {
if (!message.trim()) return;
const prompt = message.trim();
if (!prompt || !sessionId || turnStreaming) return;
setSessionError(null);
try {
await getClient().postMessage(sessionId, { message });
setMessage("");
setMessage("");
if (streamMode === "turn") {
await startTurnStream(prompt);
return;
}
try {
await getClient().postMessage(sessionId, { message: prompt });
if (!polling) {
if (streamMode === "poll") {
startPolling();
@ -266,6 +300,7 @@ export default function App() {
};
const selectSession = (session: SessionInfo) => {
stopTurnStream();
setSessionId(session.sessionId);
setAgentId(session.agent);
setAgentMode(session.agentMode);
@ -278,7 +313,12 @@ export default function App() {
setSessionError(null);
};
const createNewSession = async () => {
const createNewSession = async (nextAgentId?: string) => {
stopTurnStream();
const selectedAgent = nextAgentId ?? agentId;
if (nextAgentId) {
setAgentId(nextAgentId);
}
const chars = "abcdefghijklmnopqrstuvwxyz0123456789";
let id = "session-";
for (let i = 0; i < 8; i++) {
@ -297,7 +337,7 @@ export default function App() {
permissionMode?: string;
model?: string;
variant?: string;
} = { agent: agentId };
} = { agent: selectedAgent };
if (agentMode) body.agentMode = agentMode;
if (permissionMode) body.permissionMode = permissionMode;
if (model) body.model = model;
@ -320,6 +360,7 @@ export default function App() {
const fetchEvents = useCallback(async () => {
if (!sessionId) return;
setEventsLoading(true);
try {
const response = await getClient().getEvents(sessionId, {
offset: offsetRef.current,
@ -330,6 +371,8 @@ export default function App() {
setEventError(null);
} catch (error) {
setEventError(getErrorMessage(error, "Unable to fetch events"));
} finally {
setEventsLoading(false);
}
}, [appendEvents, getClient, sessionId]);
@ -394,6 +437,48 @@ export default function App() {
setPolling(false);
};
const startTurnStream = async (prompt: string) => {
stopPolling();
stopSse();
if (turnAbortRef.current) return;
if (!sessionId) {
setEventError("Select or create a session first.");
return;
}
setEventError(null);
setTurnStreaming(true);
const controller = new AbortController();
turnAbortRef.current = controller;
try {
for await (const event of getClient().streamTurn(
sessionId,
{ message: prompt },
undefined,
controller.signal
)) {
appendEvents([event]);
}
} catch (error) {
if (controller.signal.aborted) {
return;
}
setEventError(getErrorMessage(error, "Turn stream error."));
} finally {
if (turnAbortRef.current === controller) {
turnAbortRef.current = null;
setTurnStreaming(false);
}
}
};
const stopTurnStream = () => {
if (turnAbortRef.current) {
turnAbortRef.current.abort();
turnAbortRef.current = null;
}
setTurnStreaming(false);
};
const resetEvents = () => {
setEvents([]);
setOffset(0);
@ -580,6 +665,7 @@ export default function App() {
return () => {
stopPolling();
stopSse();
stopTurnStream();
};
}, []);
@ -604,6 +690,7 @@ export default function App() {
useEffect(() => {
if (!connected || !sessionId || polling) return;
if (streamMode === "turn") return;
const hasSession = sessions.some((session) => session.sessionId === sessionId);
if (!hasSession) return;
if (streamMode === "poll") {
@ -613,6 +700,15 @@ export default function App() {
}
}, [connected, sessionId, polling, streamMode, sessions]);
useEffect(() => {
if (streamMode === "turn") {
stopPolling();
stopSse();
} else if (turnStreaming) {
stopTurnStream();
}
}, [streamMode, turnStreaming]);
useEffect(() => {
messagesEndRef.current?.scrollIntoView({ behavior: "smooth" });
}, [transcriptEntries]);
@ -633,6 +729,16 @@ export default function App() {
const availableAgents = agents.length ? agents.map((agent) => agent.id) : defaultAgents;
const currentAgent = agents.find((agent) => agent.id === agentId);
const activeModes = modesByAgent[agentId] ?? [];
const modesLoading = modesLoadingByAgent[agentId] ?? false;
const modesError = modesErrorByAgent[agentId] ?? null;
const agentDisplayNames: Record<string, string> = {
claude: "Claude Code",
codex: "Codex",
opencode: "OpenCode",
amp: "Amp",
mock: "Mock"
};
const agentLabel = agentDisplayNames[agentId] ?? agentId;
const handleKeyDown = (event: React.KeyboardEvent<HTMLTextAreaElement>) => {
if (event.key === "Enter" && !event.shiftKey) {
@ -642,6 +748,9 @@ export default function App() {
};
const toggleStream = () => {
if (streamMode === "turn") {
return;
}
if (polling) {
if (streamMode === "poll") {
stopPolling();
@ -695,11 +804,17 @@ export default function App() {
onSelectSession={selectSession}
onRefresh={fetchSessions}
onCreateSession={createNewSession}
availableAgents={availableAgents}
agentsLoading={agentsLoading}
agentsError={agentsError}
sessionsLoading={sessionsLoading}
sessionsError={sessionsError}
/>
<ChatPanel
sessionId={sessionId}
polling={polling}
turnStreaming={turnStreaming}
transcriptEntries={transcriptEntries}
sessionError={sessionError}
message={message}
@ -708,22 +823,24 @@ export default function App() {
onKeyDown={handleKeyDown}
onCreateSession={createNewSession}
messagesEndRef={messagesEndRef}
agentId={agentId}
agentLabel={agentLabel}
agentMode={agentMode}
permissionMode={permissionMode}
model={model}
variant={variant}
streamMode={streamMode}
availableAgents={availableAgents}
activeModes={activeModes}
currentAgentVersion={currentAgent?.version ?? null}
onAgentChange={setAgentId}
modesLoading={modesLoading}
modesError={modesError}
onAgentModeChange={setAgentMode}
onPermissionModeChange={setPermissionMode}
onModelChange={setModel}
onVariantChange={setVariant}
onStreamModeChange={setStreamMode}
onToggleStream={toggleStream}
hasSession={Boolean(sessionId)}
eventError={eventError}
questionRequests={questionRequests}
permissionRequests={permissionRequests}
questionSelections={questionSelections}
@ -740,6 +857,8 @@ export default function App() {
offset={offset}
onFetchEvents={fetchEvents}
onResetEvents={resetEvents}
eventsLoading={eventsLoading}
eventsError={eventError}
requestLog={requestLog}
copiedLogId={copiedLogId}
onClearRequestLog={() => setRequestLog([])}
@ -749,6 +868,8 @@ export default function App() {
modesByAgent={modesByAgent}
onRefreshAgents={refreshAgents}
onInstallAgent={installAgent}
agentsLoading={agentsLoading}
agentsError={agentsError}
/>
</main>
</div>

View file

@ -1,4 +1,5 @@
import { Plus, RefreshCw } from "lucide-react";
import { useEffect, useRef, useState } from "react";
import type { SessionInfo } from "sandbox-agent";
const SessionSidebar = ({
@ -6,14 +7,47 @@ const SessionSidebar = ({
selectedSessionId,
onSelectSession,
onRefresh,
onCreateSession
onCreateSession,
availableAgents,
agentsLoading,
agentsError,
sessionsLoading,
sessionsError
}: {
sessions: SessionInfo[];
selectedSessionId: string;
onSelectSession: (session: SessionInfo) => void;
onRefresh: () => void;
onCreateSession: () => void;
onCreateSession: (agentId: string) => void;
availableAgents: string[];
agentsLoading: boolean;
agentsError: string | null;
sessionsLoading: boolean;
sessionsError: string | null;
}) => {
const [showMenu, setShowMenu] = useState(false);
const menuRef = useRef<HTMLDivElement | null>(null);
useEffect(() => {
if (!showMenu) return;
const handler = (event: MouseEvent) => {
if (!menuRef.current) return;
if (!menuRef.current.contains(event.target as Node)) {
setShowMenu(false);
}
};
document.addEventListener("mousedown", handler);
return () => document.removeEventListener("mousedown", handler);
}, [showMenu]);
const agentLabels: Record<string, string> = {
claude: "Claude Code",
codex: "Codex",
opencode: "OpenCode",
amp: "Amp",
mock: "Mock"
};
return (
<div className="session-sidebar">
<div className="sidebar-header">
@ -22,14 +56,46 @@ const SessionSidebar = ({
<button className="sidebar-icon-btn" onClick={onRefresh} title="Refresh sessions">
<RefreshCw size={14} />
</button>
<button className="sidebar-add-btn" onClick={onCreateSession} title="New session">
<Plus size={14} />
</button>
<div className="sidebar-add-menu-wrapper" ref={menuRef}>
<button
className="sidebar-add-btn"
onClick={() => setShowMenu((value) => !value)}
title="New session"
>
<Plus size={14} />
</button>
{showMenu && (
<div className="sidebar-add-menu">
{agentsLoading && <div className="sidebar-add-status">Loading agents...</div>}
{agentsError && <div className="sidebar-add-status error">{agentsError}</div>}
{!agentsLoading && !agentsError && availableAgents.length === 0 && (
<div className="sidebar-add-status">No agents available.</div>
)}
{!agentsLoading && !agentsError &&
availableAgents.map((id) => (
<button
key={id}
className="sidebar-add-option"
onClick={() => {
onCreateSession(id);
setShowMenu(false);
}}
>
{agentLabels[id] ?? id}
</button>
))}
</div>
)}
</div>
</div>
</div>
<div className="session-list">
{sessions.length === 0 ? (
{sessionsLoading ? (
<div className="sidebar-empty">Loading sessions...</div>
) : sessionsError ? (
<div className="sidebar-empty error">{sessionsError}</div>
) : sessions.length === 0 ? (
<div className="sidebar-empty">No sessions yet.</div>
) : (
sessions.map((session) => (

View file

@ -5,10 +5,12 @@ import type { TimelineEntry } from "./types";
const ChatMessages = ({
entries,
sessionError,
eventError,
messagesEndRef
}: {
entries: TimelineEntry[];
sessionError: string | null;
eventError: string | null;
messagesEndRef: React.RefObject<HTMLDivElement>;
}) => {
return (
@ -67,6 +69,7 @@ const ChatMessages = ({
);
})}
{sessionError && <div className="message-error">{sessionError}</div>}
{eventError && <div className="message-error">{eventError}</div>}
<div ref={messagesEndRef} />
</div>
);

View file

@ -1,4 +1,4 @@
import { MessageSquare, Plus, Terminal } from "lucide-react";
import { MessageSquare, PauseCircle, PlayCircle, Plus, Terminal } from "lucide-react";
import type { AgentModeInfo, PermissionEventData, QuestionEventData } from "sandbox-agent";
import ApprovalsTab from "../debug/ApprovalsTab";
import ChatInput from "./ChatInput";
@ -9,6 +9,7 @@ import type { TimelineEntry } from "./types";
const ChatPanel = ({
sessionId,
polling,
turnStreaming,
transcriptEntries,
sessionError,
message,
@ -17,22 +18,24 @@ const ChatPanel = ({
onKeyDown,
onCreateSession,
messagesEndRef,
agentId,
agentLabel,
agentMode,
permissionMode,
model,
variant,
streamMode,
availableAgents,
activeModes,
currentAgentVersion,
onAgentChange,
hasSession,
modesLoading,
modesError,
onAgentModeChange,
onPermissionModeChange,
onModelChange,
onVariantChange,
onStreamModeChange,
onToggleStream,
eventError,
questionRequests,
permissionRequests,
questionSelections,
@ -43,6 +46,7 @@ const ChatPanel = ({
}: {
sessionId: string;
polling: boolean;
turnStreaming: boolean;
transcriptEntries: TimelineEntry[];
sessionError: string | null;
message: string;
@ -51,22 +55,24 @@ const ChatPanel = ({
onKeyDown: (event: React.KeyboardEvent<HTMLTextAreaElement>) => void;
onCreateSession: () => void;
messagesEndRef: React.RefObject<HTMLDivElement>;
agentId: string;
agentLabel: string;
agentMode: string;
permissionMode: string;
model: string;
variant: string;
streamMode: "poll" | "sse";
availableAgents: string[];
streamMode: "poll" | "sse" | "turn";
activeModes: AgentModeInfo[];
currentAgentVersion?: string | null;
onAgentChange: (value: string) => void;
hasSession: boolean;
modesLoading: boolean;
modesError: string | null;
onAgentModeChange: (value: string) => void;
onPermissionModeChange: (value: string) => void;
onModelChange: (value: string) => void;
onVariantChange: (value: string) => void;
onStreamModeChange: (value: "poll" | "sse") => void;
onStreamModeChange: (value: "poll" | "sse" | "turn") => void;
onToggleStream: () => void;
eventError: string | null;
questionRequests: QuestionEventData[];
permissionRequests: PermissionEventData[];
questionSelections: Record<string, string[][]>;
@ -76,16 +82,57 @@ const ChatPanel = ({
onReplyPermission: (requestId: string, reply: "once" | "always" | "reject") => void;
}) => {
const hasApprovals = questionRequests.length > 0 || permissionRequests.length > 0;
const isTurnMode = streamMode === "turn";
const isStreaming = isTurnMode ? turnStreaming : polling;
const turnLabel = turnStreaming ? "Streaming" : "On Send";
return (
<div className="chat-panel">
<div className="panel-header">
<div className="panel-header-left">
<MessageSquare className="button-icon" />
<span className="panel-title">Session</span>
<span className="panel-title">{sessionId ? "Session" : "No Session"}</span>
{sessionId && <span className="session-id-display">{sessionId}</span>}
{sessionId && <span className="session-agent-display">{agentLabel}</span>}
</div>
<div className="panel-header-right">
<div className="setup-stream">
<select
className="setup-select-small"
value={streamMode}
onChange={(e) => onStreamModeChange(e.target.value as "poll" | "sse" | "turn")}
title="Stream Mode"
disabled={!sessionId}
>
<option value="poll">Poll</option>
<option value="sse">SSE</option>
<option value="turn">Turn</option>
</select>
<button
className={`setup-stream-btn ${isStreaming ? "active" : ""}`}
onClick={onToggleStream}
title={isTurnMode ? "Turn streaming starts on send" : polling ? "Stop streaming" : "Start streaming"}
disabled={!sessionId || isTurnMode}
>
{isTurnMode ? (
<>
<PlayCircle size={14} />
<span>{turnLabel}</span>
</>
) : polling ? (
<>
<PauseCircle size={14} />
<span>Pause</span>
</>
) : (
<>
<PlayCircle size={14} />
<span>Resume</span>
</>
)}
</button>
</div>
</div>
{polling && <span className="pill accent">Live</span>}
</div>
<div className="messages-container">
@ -109,6 +156,7 @@ const ChatPanel = ({
<ChatMessages
entries={transcriptEntries}
sessionError={sessionError}
eventError={eventError}
messagesEndRef={messagesEndRef}
/>
)}
@ -135,27 +183,24 @@ const ChatPanel = ({
onSendMessage={onSendMessage}
onKeyDown={onKeyDown}
placeholder={sessionId ? "Send a message..." : "Select or create a session first"}
disabled={!sessionId}
disabled={!sessionId || turnStreaming}
/>
<ChatSetup
agentId={agentId}
agentLabel={agentLabel}
agentMode={agentMode}
permissionMode={permissionMode}
model={model}
variant={variant}
streamMode={streamMode}
polling={polling}
availableAgents={availableAgents}
activeModes={activeModes}
currentAgentVersion={currentAgentVersion}
onAgentChange={onAgentChange}
modesLoading={modesLoading}
modesError={modesError}
onAgentModeChange={onAgentModeChange}
onPermissionModeChange={onPermissionModeChange}
onModelChange={onModelChange}
onVariantChange={onVariantChange}
onStreamModeChange={onStreamModeChange}
onToggleStream={onToggleStream}
hasSession={hasSession}
/>
</div>
);

View file

@ -1,60 +1,53 @@
import { PauseCircle, PlayCircle } from "lucide-react";
import type { AgentModeInfo } from "sandbox-agent";
const ChatSetup = ({
agentId,
agentLabel,
agentMode,
permissionMode,
model,
variant,
streamMode,
polling,
availableAgents,
activeModes,
currentAgentVersion,
onAgentChange,
hasSession,
modesLoading,
modesError,
onAgentModeChange,
onPermissionModeChange,
onModelChange,
onVariantChange,
onStreamModeChange,
onToggleStream
onVariantChange
}: {
agentId: string;
agentLabel: string;
agentMode: string;
permissionMode: string;
model: string;
variant: string;
streamMode: "poll" | "sse";
polling: boolean;
availableAgents: string[];
activeModes: AgentModeInfo[];
currentAgentVersion?: string | null;
onAgentChange: (value: string) => void;
hasSession: boolean;
modesLoading: boolean;
modesError: string | null;
onAgentModeChange: (value: string) => void;
onPermissionModeChange: (value: string) => void;
onModelChange: (value: string) => void;
onVariantChange: (value: string) => void;
onStreamModeChange: (value: "poll" | "sse") => void;
onToggleStream: () => void;
}) => {
const agentVersionLabel = currentAgentVersion
? `${agentLabel} v${currentAgentVersion}`
: agentLabel;
return (
<div className="setup-row">
<select className="setup-select" value={agentId} onChange={(e) => onAgentChange(e.target.value)} title="Agent">
{availableAgents.map((id) => (
<option key={id} value={id}>
{id}
</option>
))}
</select>
<select
className="setup-select"
value={agentMode}
onChange={(e) => onAgentModeChange(e.target.value)}
title="Mode"
disabled={!hasSession || modesLoading || Boolean(modesError)}
>
{activeModes.length > 0 ? (
{modesLoading ? (
<option value="">Loading modes...</option>
) : modesError ? (
<option value="">{modesError}</option>
) : activeModes.length > 0 ? (
activeModes.map((mode) => (
<option key={mode.id} value={mode.id}>
{mode.name || mode.id}
@ -70,6 +63,7 @@ const ChatSetup = ({
value={permissionMode}
onChange={(e) => onPermissionModeChange(e.target.value)}
title="Permission Mode"
disabled={!hasSession}
>
<option value="default">Default</option>
<option value="plan">Plan</option>
@ -82,6 +76,7 @@ const ChatSetup = ({
onChange={(e) => onModelChange(e.target.value)}
placeholder="Model"
title="Model"
disabled={!hasSession}
/>
<input
@ -90,40 +85,12 @@ const ChatSetup = ({
onChange={(e) => onVariantChange(e.target.value)}
placeholder="Variant"
title="Variant"
disabled={!hasSession}
/>
<div className="setup-stream">
<select
className="setup-select-small"
value={streamMode}
onChange={(e) => onStreamModeChange(e.target.value as "poll" | "sse")}
title="Stream Mode"
>
<option value="poll">Poll</option>
<option value="sse">SSE</option>
</select>
<button
className={`setup-stream-btn ${polling ? "active" : ""}`}
onClick={onToggleStream}
title={polling ? "Stop streaming" : "Start streaming"}
>
{polling ? (
<>
<PauseCircle size={14} />
<span>Pause</span>
</>
) : (
<>
<PlayCircle size={14} />
<span>Resume</span>
</>
)}
</button>
</div>
{currentAgentVersion && (
<span className="setup-version" title="Installed version">
v{currentAgentVersion}
{hasSession && (
<span className="setup-version" title="Session agent">
{agentVersionLabel}
</span>
)}
</div>

View file

@ -8,23 +8,31 @@ const AgentsTab = ({
defaultAgents,
modesByAgent,
onRefresh,
onInstall
onInstall,
loading,
error
}: {
agents: AgentInfo[];
defaultAgents: string[];
modesByAgent: Record<string, AgentModeInfo[]>;
onRefresh: () => void;
onInstall: (agentId: string, reinstall: boolean) => void;
loading: boolean;
error: string | null;
}) => {
return (
<>
<div className="inline-row" style={{ marginBottom: 16 }}>
<button className="button secondary small" onClick={onRefresh}>
<button className="button secondary small" onClick={onRefresh} disabled={loading}>
<RefreshCw className="button-icon" /> Refresh
</button>
</div>
{agents.length === 0 && <div className="card-meta">No agents reported. Click refresh to check.</div>}
{error && <div className="banner error">{error}</div>}
{loading && <div className="card-meta">Loading agents...</div>}
{!loading && agents.length === 0 && (
<div className="card-meta">No agents reported. Click refresh to check.</div>
)}
{(agents.length
? agents

View file

@ -14,6 +14,8 @@ const DebugPanel = ({
offset,
onFetchEvents,
onResetEvents,
eventsLoading,
eventsError,
requestLog,
copiedLogId,
onClearRequestLog,
@ -22,7 +24,9 @@ const DebugPanel = ({
defaultAgents,
modesByAgent,
onRefreshAgents,
onInstallAgent
onInstallAgent,
agentsLoading,
agentsError
}: {
debugTab: DebugTab;
onDebugTabChange: (tab: DebugTab) => void;
@ -30,6 +34,8 @@ const DebugPanel = ({
offset: number;
onFetchEvents: () => void;
onResetEvents: () => void;
eventsLoading: boolean;
eventsError: string | null;
requestLog: RequestLog[];
copiedLogId: number | null;
onClearRequestLog: () => void;
@ -39,6 +45,8 @@ const DebugPanel = ({
modesByAgent: Record<string, AgentModeInfo[]>;
onRefreshAgents: () => void;
onInstallAgent: (agentId: string, reinstall: boolean) => void;
agentsLoading: boolean;
agentsError: string | null;
}) => {
return (
<div className="debug-panel">
@ -69,7 +77,14 @@ const DebugPanel = ({
)}
{debugTab === "events" && (
<EventsTab events={events} offset={offset} onFetch={onFetchEvents} onClear={onResetEvents} />
<EventsTab
events={events}
offset={offset}
onFetch={onFetchEvents}
onClear={onResetEvents}
loading={eventsLoading}
error={eventsError}
/>
)}
{debugTab === "agents" && (
@ -79,6 +94,8 @@ const DebugPanel = ({
modesByAgent={modesByAgent}
onRefresh={onRefreshAgents}
onInstall={onInstallAgent}
loading={agentsLoading}
error={agentsError}
/>
)}
</div>

View file

@ -8,12 +8,16 @@ const EventsTab = ({
events,
offset,
onFetch,
onClear
onClear,
loading,
error
}: {
events: UniversalEvent[];
offset: number;
onFetch: () => void;
onClear: () => void;
loading: boolean;
error: string | null;
}) => {
const [collapsedEvents, setCollapsedEvents] = useState<Record<string, boolean>>({});
@ -28,8 +32,8 @@ const EventsTab = ({
<div className="inline-row" style={{ marginBottom: 12, justifyContent: "space-between" }}>
<span className="card-meta">Offset: {offset}</span>
<div className="inline-row">
<button className="button ghost small" onClick={onFetch}>
Fetch
<button className="button ghost small" onClick={onFetch} disabled={loading}>
{loading ? "Loading..." : "Fetch"}
</button>
<button className="button ghost small" onClick={onClear}>
Clear
@ -37,8 +41,12 @@ const EventsTab = ({
</div>
</div>
{error && <div className="banner error">{error}</div>}
{events.length === 0 ? (
<div className="card-meta">No events yet. Start streaming to receive events.</div>
<div className="card-meta">
{loading ? "Loading events..." : "No events yet. Start streaming to receive events."}
</div>
) : (
<div className="event-list">
{[...events].reverse().map((event) => {

View file

@ -254,6 +254,70 @@ Codex output is converted via `convertCodexOutput()`:
- Use `resumeThread(threadId)` to continue conversation
- Thread ID is captured from `thread.started` event or thread object
## Shared App-Server Architecture (Daemon Implementation)
The sandbox daemon uses a **single shared Codex app-server process** to handle multiple sessions, similar to OpenCode's server model. This differs from Claude/Amp which spawn a new process per turn.
### Architecture Comparison
| Agent | Model | Process Lifetime | Session ID |
|-------|-------|------------------|------------|
| Claude | Subprocess | Per-turn (killed on TurnCompleted) | `--resume` flag |
| Amp | Subprocess | Per-turn | `--continue` flag |
| OpenCode | HTTP Server | Daemon lifetime | Session ID via API |
| **Codex** | **Stdio Server** | **Daemon lifetime** | **Thread ID via JSON-RPC** |
### Daemon Flow
1. **First Codex session created**: Spawns `codex app-server` process, performs `initialize`/`initialized` handshake
2. **Session creation**: Sends `thread/start` request, captures `thread_id` as `native_session_id`
3. **Message sent**: Sends `turn/start` request with `thread_id`, streams notifications back to session
4. **Multi-turn**: Reuses same `thread_id`, process stays alive, no respawn needed
5. **Daemon shutdown**: Process terminated with daemon
### Why This Approach?
1. **Performance**: No process spawn overhead per message
2. **Multi-turn support**: Thread persists in server memory, no resume needed
3. **Consistent with OpenCode**: Similar server-based pattern reduces code complexity
4. **API alignment**: Matches Codex's intended app-server usage pattern
### Protocol Details
The shared server uses JSON-RPC 2.0 for request/response correlation:
```
Daemon Codex App-Server
| |
|-- initialize {id: 1} ------------>|
|<-- response {id: 1} --------------|
|-- initialized (notification) ---->|
| |
|-- thread/start {id: 2} ---------->|
|<-- response {id: 2, thread.id} ---|
|<-- thread/started (notification) -|
| |
|-- turn/start {id: 3, threadId} -->|
|<-- turn/started (notification) ---|
|<-- item/* (notifications) --------|
|<-- turn/completed (notification) -|
```
### Thread-to-Session Routing
Notifications are routed to the correct session by extracting `threadId` from each notification:
```rust
fn codex_thread_id_from_server_notification(notification) -> Option<String> {
// All thread-scoped notifications include threadId field
match notification {
TurnStarted(params) => Some(params.thread_id),
ItemCompleted(params) => Some(params.thread_id),
// ... etc
}
}
```
## Notes
- SDK is dynamically imported to reduce bundle size

View file

@ -148,6 +148,19 @@ function isStable(version: string) {
return parseSemver(version).prerelease.length === 0;
}
function getNpmTag(version: string, latest: boolean) {
if (latest) return null;
const prerelease = parseSemver(version).prerelease;
if (prerelease.length === 0) {
return "next";
}
const hasRc = prerelease.some((part) => part.toLowerCase().startsWith("rc"));
if (hasRc) {
return "rc";
}
throw new Error(`Prerelease versions must use rc tag when not latest: ${version}`);
}
function getAllGitVersions() {
try {
execFileSync("git", ["fetch", "--tags", "--force", "--quiet"], {
@ -411,18 +424,22 @@ function publishCrates(rootDir: string, version: string) {
}
}
function publishNpmSdk(rootDir: string, version: string) {
function publishNpmSdk(rootDir: string, version: string, latest: boolean) {
const sdkDir = path.join(rootDir, "sdks", "typescript");
console.log("==> Publishing TypeScript SDK to npm");
const npmTag = getNpmTag(version, latest);
run("npm", ["version", version, "--no-git-tag-version", "--allow-same-version"], { cwd: sdkDir });
run("pnpm", ["install"], { cwd: sdkDir });
run("pnpm", ["run", "build"], { cwd: sdkDir });
run("npm", ["publish", "--access", "public"], { cwd: sdkDir });
const publishArgs = ["publish", "--access", "public"];
if (npmTag) publishArgs.push("--tag", npmTag);
run("npm", publishArgs, { cwd: sdkDir });
}
function publishNpmCli(rootDir: string, version: string) {
function publishNpmCli(rootDir: string, version: string, latest: boolean) {
const cliDir = path.join(rootDir, "sdks", "cli");
const distDir = path.join(rootDir, "dist");
const npmTag = getNpmTag(version, latest);
for (const [target, info] of Object.entries(PLATFORM_MAP)) {
const platformDir = path.join(cliDir, "platforms", info.pkg);
@ -436,7 +453,9 @@ function publishNpmCli(rootDir: string, version: string) {
console.log(`==> Publishing @sandbox-agent/cli-${info.pkg}`);
run("npm", ["version", version, "--no-git-tag-version", "--allow-same-version"], { cwd: platformDir });
run("npm", ["publish", "--access", "public"], { cwd: platformDir });
const publishArgs = ["publish", "--access", "public"];
if (npmTag) publishArgs.push("--tag", npmTag);
run("npm", publishArgs, { cwd: platformDir });
}
console.log("==> Publishing @sandbox-agent/cli");
@ -447,7 +466,9 @@ function publishNpmCli(rootDir: string, version: string) {
pkg.optionalDependencies[dep] = version;
}
fs.writeFileSync(pkgPath, JSON.stringify(pkg, null, 2) + "\n");
run("npm", ["publish", "--access", "public"], { cwd: cliDir });
const publishArgs = ["publish", "--access", "public"];
if (npmTag) publishArgs.push("--tag", npmTag);
run("npm", publishArgs, { cwd: cliDir });
}
function validateGit(rootDir: string) {
@ -542,10 +563,10 @@ async function main() {
publishCrates(rootDir, version);
}
if (flags.has("--publish-npm-sdk")) {
publishNpmSdk(rootDir, version);
publishNpmSdk(rootDir, version, latest);
}
if (flags.has("--publish-npm-cli")) {
publishNpmCli(rootDir, version);
publishNpmCli(rootDir, version, latest);
}
if (flags.has("--upload-typescript")) {
uploadTypescriptArtifacts(rootDir, version, latest);
@ -626,11 +647,11 @@ async function main() {
}
if (shouldRun("publish-npm-sdk")) {
publishNpmSdk(rootDir, version);
publishNpmSdk(rootDir, version, latest);
}
if (shouldRun("publish-npm-cli")) {
publishNpmCli(rootDir, version);
publishNpmCli(rootDir, version, latest);
}
if (shouldRun("upload-typescript")) {

View file

@ -13,6 +13,7 @@ import type {
ProblemDetails,
QuestionReplyRequest,
SessionListResponse,
TurnStreamQuery,
UniversalEvent,
} from "./types.ts";
@ -142,45 +143,37 @@ export class SandboxAgent {
});
}
async postMessageStream(
sessionId: string,
request: MessageRequest,
query?: TurnStreamQuery,
signal?: AbortSignal,
): Promise<Response> {
return this.requestRaw("POST", `${API_PREFIX}/sessions/${encodeURIComponent(sessionId)}/messages/stream`, {
query,
body: request,
accept: "text/event-stream",
signal,
});
}
async *streamEvents(
sessionId: string,
query?: EventsQuery,
signal?: AbortSignal,
): AsyncGenerator<UniversalEvent, void, void> {
const response = await this.getEventsSse(sessionId, query, signal);
if (!response.body) {
throw new Error("SSE stream is not readable in this environment.");
}
yield* this.parseSseStream(response);
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) {
break;
}
// Normalize CRLF to LF for consistent parsing
buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, "\n");
let index = buffer.indexOf("\n\n");
while (index !== -1) {
const chunk = buffer.slice(0, index);
buffer = buffer.slice(index + 2);
const dataLines = chunk
.split("\n")
.filter((line) => line.startsWith("data:"));
if (dataLines.length > 0) {
const payload = dataLines
.map((line) => line.slice(5).trim())
.join("\n");
if (payload) {
yield JSON.parse(payload) as UniversalEvent;
}
}
index = buffer.indexOf("\n\n");
}
}
async *streamTurn(
sessionId: string,
request: MessageRequest,
query?: TurnStreamQuery,
signal?: AbortSignal,
): AsyncGenerator<UniversalEvent, void, void> {
const response = await this.postMessageStream(sessionId, request, query, signal);
yield* this.parseSseStream(response);
}
async replyQuestion(
@ -297,6 +290,42 @@ export class SandboxAgent {
return undefined;
}
}
private async *parseSseStream(response: Response): AsyncGenerator<UniversalEvent, void, void> {
if (!response.body) {
throw new Error("SSE stream is not readable in this environment.");
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) {
break;
}
// Normalize CRLF to LF for consistent parsing
buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, "\n");
let index = buffer.indexOf("\n\n");
while (index !== -1) {
const chunk = buffer.slice(0, index);
buffer = buffer.slice(index + 2);
const dataLines = chunk
.split("\n")
.filter((line) => line.startsWith("data:"));
if (dataLines.length > 0) {
const payload = dataLines
.map((line) => line.slice(5).trim())
.join("\n");
if (payload) {
yield JSON.parse(payload) as UniversalEvent;
}
}
index = buffer.indexOf("\n\n");
}
}
}
}
const normalizeSpawnOptions = (

View file

@ -32,6 +32,9 @@ export interface paths {
"/v1/sessions/{session_id}/messages": {
post: operations["post_message"];
};
"/v1/sessions/{session_id}/messages/stream": {
post: operations["post_message_stream"];
};
"/v1/sessions/{session_id}/permissions/{permission_id}/reply": {
post: operations["reply_permission"];
};
@ -258,6 +261,9 @@ export interface components {
};
/** @enum {string} */
TerminatedBy: "agent" | "daemon";
TurnStreamQuery: {
includeRaw?: boolean | null;
};
UniversalEvent: {
data: components["schemas"]["UniversalEventData"];
event_id: string;
@ -480,6 +486,34 @@ export interface operations {
};
};
};
post_message_stream: {
parameters: {
query?: {
/** @description Include raw provider payloads */
include_raw?: boolean | null;
};
path: {
/** @description Session id */
session_id: string;
};
};
requestBody: {
content: {
"application/json": components["schemas"]["MessageRequest"];
};
};
responses: {
/** @description SSE event stream */
200: {
content: never;
};
404: {
content: {
"application/json": components["schemas"]["ProblemDetails"];
};
};
};
};
reply_permission: {
parameters: {
path: {

View file

@ -41,6 +41,7 @@ export type {
SessionListResponse,
SessionStartedData,
TerminatedBy,
TurnStreamQuery,
UniversalEvent,
UniversalEventData,
UniversalEventType,

View file

@ -39,6 +39,7 @@ export type SessionInfo = S["SessionInfo"];
export type SessionListResponse = S["SessionListResponse"];
export type SessionStartedData = S["SessionStartedData"];
export type TerminatedBy = S["TerminatedBy"];
export type TurnStreamQuery = S["TurnStreamQuery"];
export type UniversalEvent = S["UniversalEvent"];
export type UniversalEventData = S["UniversalEventData"];
export type UniversalEventType = S["UniversalEventType"];

View file

@ -164,6 +164,31 @@ describe("SandboxAgent", () => {
});
});
describe("postMessageStream", () => {
it("posts message and requests SSE", async () => {
const mockFetch = vi.fn().mockResolvedValue(
new Response("", {
status: 200,
headers: { "Content-Type": "text/event-stream" },
})
);
const client = await SandboxAgent.connect({
baseUrl: "http://localhost:8080",
fetch: mockFetch,
});
await client.postMessageStream("test-session", { message: "Hello" }, { includeRaw: true });
expect(mockFetch).toHaveBeenCalledWith(
"http://localhost:8080/v1/sessions/test-session/messages/stream?includeRaw=true",
expect.objectContaining({
method: "POST",
body: JSON.stringify({ message: "Hello" }),
})
);
});
});
describe("getEvents", () => {
it("returns events", async () => {
const events = { events: [], hasMore: false };

View file

@ -122,6 +122,9 @@ enum SessionsCommand {
#[command(name = "send-message")]
/// Send a message to an existing session.
SendMessage(SessionMessageArgs),
#[command(name = "send-message-stream")]
/// Send a message and stream the response for one turn.
SendMessageStream(SessionMessageStreamArgs),
#[command(name = "terminate")]
/// Terminate a session.
Terminate(SessionTerminateArgs),
@ -195,6 +198,17 @@ struct SessionMessageArgs {
client: ClientArgs,
}
#[derive(Args, Debug)]
struct SessionMessageStreamArgs {
session_id: String,
#[arg(long, short = 'm')]
message: String,
#[arg(long)]
include_raw: bool,
#[command(flatten)]
client: ClientArgs,
}
#[derive(Args, Debug)]
struct SessionEventsArgs {
session_id: String,
@ -443,6 +457,22 @@ fn run_sessions(command: &SessionsCommand, cli: &Cli) -> Result<(), CliError> {
let response = ctx.post(&path, &body)?;
print_empty_response(response)
}
SessionsCommand::SendMessageStream(args) => {
let ctx = ClientContext::new(cli, &args.client)?;
let body = MessageRequest {
message: args.message.clone(),
};
let path = format!("{API_PREFIX}/sessions/{}/messages/stream", args.session_id);
let response = ctx.post_with_query(
&path,
&body,
&[(
"include_raw",
if args.include_raw { Some("true".to_string()) } else { None },
)],
)?;
print_text_response(response)
}
SessionsCommand::Terminate(args) => {
let ctx = ClientContext::new(cli, &args.client)?;
let path = format!("{API_PREFIX}/sessions/{}/terminate", args.session_id);
@ -850,6 +880,21 @@ impl ClientContext {
Ok(self.request(Method::POST, path).json(body).send()?)
}
fn post_with_query<T: Serialize>(
&self,
path: &str,
body: &T,
query: &[(&str, Option<String>)],
) -> Result<reqwest::blocking::Response, CliError> {
let mut request = self.request(Method::POST, path).json(body);
for (key, value) in query {
if let Some(value) = value {
request = request.query(&[(key, value)]);
}
}
Ok(request.send()?)
}
fn post_empty(&self, path: &str) -> Result<reqwest::blocking::Response, CliError> {
Ok(self.request(Method::POST, path).send()?)
}

File diff suppressed because it is too large Load diff

View file

@ -8,11 +8,15 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use reqwest::Client;
use serde::Serialize;
use time::OffsetDateTime;
use tokio::time::Instant;
const TELEMETRY_URL: &str = "https://tc.rivet.dev";
const TELEMETRY_ENV_DEBUG: &str = "SANDBOX_AGENT_TELEMETRY_DEBUG";
const TELEMETRY_ID_FILE: &str = "telemetry_id";
const TELEMETRY_TIMEOUT_MS: u64 = 800;
const TELEMETRY_LAST_SENT_FILE: &str = "telemetry_last_sent";
const TELEMETRY_TIMEOUT_MS: u64 = 2_000;
const TELEMETRY_INTERVAL_SECS: u64 = 300;
const TELEMETRY_MIN_GAP_SECS: i64 = 300;
#[derive(Debug, Serialize)]
struct TelemetryEvent {
@ -49,7 +53,6 @@ struct OsInfo {
#[derive(Debug, Serialize)]
struct ProviderInfo {
name: String,
confidence: String,
#[serde(skip_serializing_if = "Option::is_none")]
method: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
@ -69,11 +72,10 @@ pub fn telemetry_enabled(no_telemetry: bool) -> bool {
}
pub fn log_enabled_message() {
tracing::info!("anonymous telemetry is enabled; disable with --no-telemetry");
tracing::info!("anonymous telemetry is enabled, disable with --no-telemetry");
}
pub fn spawn_telemetry_task() {
let event = build_event();
tokio::spawn(async move {
let client = match Client::builder()
.timeout(Duration::from_millis(TELEMETRY_TIMEOUT_MS))
@ -86,21 +88,38 @@ pub fn spawn_telemetry_task() {
}
};
if let Err(err) = client.post(TELEMETRY_URL).json(&event).send().await {
tracing::debug!(error = %err, "telemetry request failed");
attempt_send(&client).await;
let start = Instant::now() + Duration::from_secs(TELEMETRY_INTERVAL_SECS);
let mut interval = tokio::time::interval_at(start, Duration::from_secs(TELEMETRY_INTERVAL_SECS));
loop {
interval.tick().await;
attempt_send(&client).await;
}
});
}
fn build_event() -> TelemetryEvent {
async fn attempt_send(client: &Client) {
let dt = OffsetDateTime::now_utc().unix_timestamp();
if !should_send(dt) {
return;
}
let event = build_event(dt);
if let Err(err) = client.post(TELEMETRY_URL).json(&event).send().await {
tracing::debug!(error = %err, "telemetry request failed");
return;
}
write_last_sent(dt);
}
fn build_event(dt: i64) -> TelemetryEvent {
let eid = load_or_create_id();
TelemetryEvent {
p: "sandbox-agent".to_string(),
dt,
et: "sandbox".to_string(),
eid,
ev: "entity_snapshot".to_string(),
ev: "entity_beacon".to_string(),
d: TelemetryData {
version: env!("CARGO_PKG_VERSION").to_string(),
os: OsInfo {
@ -138,9 +157,46 @@ fn load_or_create_id() -> String {
}
fn telemetry_id_path() -> PathBuf {
telemetry_dir().join(TELEMETRY_ID_FILE)
}
fn telemetry_last_sent_path() -> PathBuf {
telemetry_dir().join(TELEMETRY_LAST_SENT_FILE)
}
fn telemetry_dir() -> PathBuf {
dirs::data_dir()
.map(|dir| dir.join("sandbox-agent").join(TELEMETRY_ID_FILE))
.unwrap_or_else(|| PathBuf::from(".sandbox-agent").join(TELEMETRY_ID_FILE))
.map(|dir| dir.join("sandbox-agent"))
.unwrap_or_else(|| PathBuf::from(".sandbox-agent"))
}
fn should_send(now: i64) -> bool {
if let Some(last) = read_last_sent() {
if now >= last && now - last < TELEMETRY_MIN_GAP_SECS {
return false;
}
}
true
}
fn read_last_sent() -> Option<i64> {
let path = telemetry_last_sent_path();
fs::read_to_string(&path)
.ok()
.and_then(|value| value.trim().parse::<i64>().ok())
}
fn write_last_sent(timestamp: i64) {
let path = telemetry_last_sent_path();
if let Some(parent) = path.parent() {
if let Err(err) = fs::create_dir_all(parent) {
tracing::debug!(error = %err, "failed to create telemetry directory");
return;
}
}
if let Ok(mut file) = fs::OpenOptions::new().create(true).write(true).truncate(true).open(&path) {
let _ = file.write_all(timestamp.to_string().as_bytes());
}
}
fn generate_id() -> String {
@ -185,7 +241,6 @@ fn detect_provider() -> ProviderInfo {
]);
return ProviderInfo {
name: "e2b".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -206,7 +261,6 @@ fn detect_provider() -> ProviderInfo {
]);
return ProviderInfo {
name: "vercel".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -219,7 +273,6 @@ fn detect_provider() -> ProviderInfo {
]);
return ProviderInfo {
name: "modal".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -232,7 +285,6 @@ fn detect_provider() -> ProviderInfo {
]);
return ProviderInfo {
name: "fly.io".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -245,7 +297,6 @@ fn detect_provider() -> ProviderInfo {
]);
return ProviderInfo {
name: "replit".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -254,7 +305,6 @@ fn detect_provider() -> ProviderInfo {
if env::var("CODESANDBOX_HOST").is_ok() || env::var("CSB_BASE_PREVIEW_HOST").is_ok() {
return ProviderInfo {
name: "codesandbox".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata: None,
};
@ -264,7 +314,6 @@ fn detect_provider() -> ProviderInfo {
let metadata = metadata_or_none([("name", env::var("CODESPACE_NAME").ok())]);
return ProviderInfo {
name: "github-codespaces".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -274,7 +323,6 @@ fn detect_provider() -> ProviderInfo {
let metadata = metadata_or_none([("environment", env::var("RAILWAY_ENVIRONMENT").ok())]);
return ProviderInfo {
name: "railway".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -284,7 +332,6 @@ fn detect_provider() -> ProviderInfo {
let metadata = metadata_or_none([("serviceId", env::var("RENDER_SERVICE_ID").ok())]);
return ProviderInfo {
name: "render".to_string(),
confidence: "high".to_string(),
method: Some("env".to_string()),
metadata,
};
@ -293,7 +340,6 @@ fn detect_provider() -> ProviderInfo {
if detect_daytona() {
return ProviderInfo {
name: "daytona".to_string(),
confidence: "medium".to_string(),
method: Some("filesystem".to_string()),
metadata: None,
};
@ -302,7 +348,6 @@ fn detect_provider() -> ProviderInfo {
if detect_docker() {
return ProviderInfo {
name: "docker".to_string(),
confidence: "high".to_string(),
method: Some("filesystem".to_string()),
metadata: None,
};
@ -310,7 +355,6 @@ fn detect_provider() -> ProviderInfo {
ProviderInfo {
name: "unknown".to_string(),
confidence: "low".to_string(),
method: None,
metadata: None,
}

View file

@ -1,657 +0,0 @@
use std::collections::HashMap;
use std::time::{Duration, Instant};
use axum::body::Body;
use axum::http::{Method, Request, StatusCode};
use axum::Router;
use http_body_util::BodyExt;
use serde_json::{json, Value};
use tempfile::TempDir;
use tower::util::ServiceExt;
use sandbox_agent_agent_management::agents::{AgentId, AgentManager};
use sandbox_agent_agent_management::testing::test_agents_from_env;
use sandbox_agent_agent_credentials::ExtractedCredentials;
use sandbox_agent::router::{
build_router,
AgentCapabilities,
AgentListResponse,
AuthConfig,
};
const PROMPT: &str = "Reply with exactly the single word OK.";
const TOOL_PROMPT: &str =
"Use the bash tool to run `ls` in the current directory. Do not answer without using the tool.";
const QUESTION_PROMPT: &str =
"Call the AskUserQuestion tool with exactly one yes/no question and wait for a reply. Do not answer yourself.";
/// Agent-agnostic event sequence tests.
///
/// These tests assert that the universal schema output is valid and consistent
/// across agents, and they use capability flags from /v1/agents to skip
/// unsupported flows.
struct TestApp {
app: Router,
_install_dir: TempDir,
}
impl TestApp {
fn new() -> Self {
let install_dir = tempfile::tempdir().expect("create temp install dir");
let manager = AgentManager::new(install_dir.path())
.expect("create agent manager");
let state = sandbox_agent::router::AppState::new(AuthConfig::disabled(), manager);
let app = build_router(state);
Self {
app,
_install_dir: install_dir,
}
}
}
struct EnvGuard {
saved: HashMap<String, Option<String>>,
}
impl Drop for EnvGuard {
fn drop(&mut self) {
for (key, value) in &self.saved {
match value {
Some(value) => std::env::set_var(key, value),
None => std::env::remove_var(key),
}
}
}
}
fn apply_credentials(creds: &ExtractedCredentials) -> EnvGuard {
let keys = ["ANTHROPIC_API_KEY", "CLAUDE_API_KEY", "OPENAI_API_KEY", "CODEX_API_KEY"];
let mut saved = HashMap::new();
for key in keys {
saved.insert(key.to_string(), std::env::var(key).ok());
}
match creds.anthropic.as_ref() {
Some(cred) => {
std::env::set_var("ANTHROPIC_API_KEY", &cred.api_key);
std::env::set_var("CLAUDE_API_KEY", &cred.api_key);
}
None => {
std::env::remove_var("ANTHROPIC_API_KEY");
std::env::remove_var("CLAUDE_API_KEY");
}
}
match creds.openai.as_ref() {
Some(cred) => {
std::env::set_var("OPENAI_API_KEY", &cred.api_key);
std::env::set_var("CODEX_API_KEY", &cred.api_key);
}
None => {
std::env::remove_var("OPENAI_API_KEY");
std::env::remove_var("CODEX_API_KEY");
}
}
EnvGuard { saved }
}
async fn send_json(
app: &Router,
method: Method,
path: &str,
body: Option<Value>,
) -> (StatusCode, Value) {
let request = Request::builder()
.method(method)
.uri(path)
.header("content-type", "application/json")
.body(Body::from(body.map(|value| value.to_string()).unwrap_or_default()))
.expect("request");
let response = app
.clone()
.oneshot(request)
.await
.expect("response");
let status = response.status();
let bytes = response
.into_body()
.collect()
.await
.expect("body")
.to_bytes();
let payload = if bytes.is_empty() {
Value::Null
} else {
serde_json::from_slice(&bytes).unwrap_or(Value::Null)
};
(status, payload)
}
async fn send_status(app: &Router, method: Method, path: &str, body: Option<Value>) -> StatusCode {
let (status, _) = send_json(app, method, path, body).await;
status
}
async fn install_agent(app: &Router, agent: AgentId) {
let status = send_status(
app,
Method::POST,
&format!("/v1/agents/{}/install", agent.as_str()),
Some(json!({})),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "install agent {}", agent.as_str());
}
async fn create_session(app: &Router, agent: AgentId, session_id: &str, permission_mode: &str) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}"),
Some(json!({
"agent": agent.as_str(),
"permissionMode": permission_mode,
})),
)
.await;
assert_eq!(status, StatusCode::OK, "create session");
}
async fn create_session_with_mode(
app: &Router,
agent: AgentId,
session_id: &str,
agent_mode: &str,
permission_mode: &str,
) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}"),
Some(json!({
"agent": agent.as_str(),
"agentMode": agent_mode,
"permissionMode": permission_mode,
})),
)
.await;
assert_eq!(status, StatusCode::OK, "create session");
}
fn test_permission_mode(agent: AgentId) -> &'static str {
match agent {
AgentId::Opencode => "default",
_ => "bypass",
}
}
async fn send_message(app: &Router, session_id: &str, message: &str) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}/messages"),
Some(json!({ "message": message })),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "send message");
}
async fn poll_events_until<F>(
app: &Router,
session_id: &str,
timeout: Duration,
mut stop: F,
) -> Vec<Value>
where
F: FnMut(&[Value]) -> bool,
{
let start = Instant::now();
let mut offset = 0u64;
let mut events = Vec::new();
while start.elapsed() < timeout {
let path = format!("/v1/sessions/{session_id}/events?offset={offset}&limit=200");
let (status, payload) = send_json(app, Method::GET, &path, None).await;
assert_eq!(status, StatusCode::OK, "poll events");
let new_events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if !new_events.is_empty() {
if let Some(last) = new_events
.last()
.and_then(|event| event.get("sequence"))
.and_then(Value::as_u64)
{
offset = last;
}
events.extend(new_events);
if stop(&events) {
break;
}
}
tokio::time::sleep(Duration::from_millis(800)).await;
}
events
}
async fn fetch_capabilities(app: &Router) -> HashMap<String, AgentCapabilities> {
let (status, payload) = send_json(app, Method::GET, "/v1/agents", None).await;
assert_eq!(status, StatusCode::OK, "list agents");
let response: AgentListResponse = serde_json::from_value(payload).expect("agents payload");
response
.agents
.into_iter()
.map(|agent| (agent.id, agent.capabilities))
.collect()
}
fn has_event_type(events: &[Value], event_type: &str) -> bool {
events
.iter()
.any(|event| event.get("type").and_then(Value::as_str) == Some(event_type))
}
fn find_assistant_message_item(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("item.completed") {
return None;
}
let item = event.get("data")?.get("item")?;
let role = item.get("role")?.as_str()?;
let kind = item.get("kind")?.as_str()?;
if role != "assistant" || kind != "message" {
return None;
}
item.get("item_id")?.as_str().map(|id| id.to_string())
})
}
fn event_sequence(event: &Value) -> Option<u64> {
event.get("sequence").and_then(Value::as_u64)
}
fn find_item_event_seq(events: &[Value], event_type: &str, item_id: &str) -> Option<u64> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some(event_type) {
return None;
}
match event_type {
"item.delta" => {
let data = event.get("data")?;
let id = data.get("item_id")?.as_str()?;
if id == item_id {
event_sequence(event)
} else {
None
}
}
_ => {
let item = event.get("data")?.get("item")?;
let id = item.get("item_id")?.as_str()?;
if id == item_id {
event_sequence(event)
} else {
None
}
}
}
})
}
fn find_permission_id(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("permission.requested") {
return None;
}
event
.get("data")
.and_then(|data| data.get("permission_id"))
.and_then(Value::as_str)
.map(|id| id.to_string())
})
}
fn find_question_id(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("question.requested") {
return None;
}
event
.get("data")
.and_then(|data| data.get("question_id"))
.and_then(Value::as_str)
.map(|id| id.to_string())
})
}
fn find_first_answer(events: &[Value]) -> Option<Vec<Vec<String>>> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("question.requested") {
return None;
}
let options = event
.get("data")
.and_then(|data| data.get("options"))
.and_then(Value::as_array)?;
let option = options.first()?.as_str()?.to_string();
Some(vec![vec![option]])
})
}
fn find_tool_call(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("item.started")
&& event.get("type").and_then(Value::as_str) != Some("item.completed")
{
return None;
}
let item = event.get("data")?.get("item")?;
let kind = item.get("kind")?.as_str()?;
if kind != "tool_call" {
return None;
}
item.get("item_id")?.as_str().map(|id| id.to_string())
})
}
fn has_tool_result(events: &[Value]) -> bool {
events.iter().any(|event| {
if event.get("type").and_then(Value::as_str) != Some("item.completed") {
return false;
}
let item = match event.get("data").and_then(|data| data.get("item")) {
Some(item) => item,
None => return false,
};
item.get("kind").and_then(Value::as_str) == Some("tool_result")
})
}
fn expect_basic_sequence(events: &[Value]) {
assert!(has_event_type(events, "session.started"), "session.started missing");
let item_id = find_assistant_message_item(events).expect("assistant message missing");
let started_seq = find_item_event_seq(events, "item.started", &item_id)
.expect("item.started missing");
// Intentionally require deltas here to validate our synthetic delta behavior.
let delta_seq = find_item_event_seq(events, "item.delta", &item_id)
.expect("item.delta missing");
let completed_seq = find_item_event_seq(events, "item.completed", &item_id)
.expect("item.completed missing");
assert!(started_seq < delta_seq, "item.started must precede delta");
assert!(delta_seq < completed_seq, "delta must precede completion");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_agnostic_basic_reply() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("basic-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, "default").await;
send_message(&app.app, &session_id, PROMPT).await;
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
has_event_type(events, "error") || find_assistant_message_item(events).is_some()
})
.await;
assert!(
!events.is_empty(),
"no events collected for {}",
config.agent.as_str()
);
expect_basic_sequence(&events);
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if caps.tool_calls {
assert!(
!events.iter().any(|event| {
event.get("type").and_then(Value::as_str) == Some("agent.unparsed")
}),
"agent.unparsed event detected"
);
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_agnostic_tool_flow() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if !caps.tool_calls {
continue;
}
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("tool-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, test_permission_mode(config.agent)).await;
send_message(&app.app, &session_id, TOOL_PROMPT).await;
let start = Instant::now();
let mut offset = 0u64;
let mut events = Vec::new();
let mut replied = false;
while start.elapsed() < Duration::from_secs(180) {
let path = format!("/v1/sessions/{session_id}/events?offset={offset}&limit=200");
let (status, payload) = send_json(&app.app, Method::GET, &path, None).await;
assert_eq!(status, StatusCode::OK, "poll events");
let new_events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if !new_events.is_empty() {
if let Some(last) = new_events
.last()
.and_then(|event| event.get("sequence"))
.and_then(Value::as_u64)
{
offset = last;
}
events.extend(new_events);
if !replied {
if let Some(permission_id) = find_permission_id(&events) {
let _ = send_status(
&app.app,
Method::POST,
&format!(
"/v1/sessions/{session_id}/permissions/{permission_id}/reply"
),
Some(json!({ "reply": "once" })),
)
.await;
replied = true;
}
}
if has_tool_result(&events) {
break;
}
}
tokio::time::sleep(Duration::from_millis(800)).await;
}
let tool_call = find_tool_call(&events);
let tool_result = has_tool_result(&events);
assert!(
tool_call.is_some(),
"tool_call missing for tool-capable agent {}",
config.agent.as_str()
);
if tool_call.is_some() {
assert!(
tool_result,
"tool_result missing after tool_call for {}",
config.agent.as_str()
);
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_agnostic_permission_flow() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if !(caps.plan_mode && caps.permissions) {
continue;
}
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("perm-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, "plan").await;
send_message(&app.app, &session_id, TOOL_PROMPT).await;
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
find_permission_id(events).is_some() || has_event_type(events, "error")
})
.await;
let permission_id = find_permission_id(&events).expect("permission.requested missing");
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/permissions/{permission_id}/reply"),
Some(json!({ "reply": "once" })),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "permission reply");
let resolved = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
events.iter().any(|event| {
event.get("type").and_then(Value::as_str) == Some("permission.resolved")
})
})
.await;
assert!(
resolved.iter().any(|event| {
event.get("type").and_then(Value::as_str) == Some("permission.resolved")
&& event
.get("synthetic")
.and_then(Value::as_bool)
.unwrap_or(false)
}),
"permission.resolved should be synthetic"
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_agnostic_question_flow() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if !caps.questions {
continue;
}
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("question-{}", config.agent.as_str());
create_session_with_mode(&app.app, config.agent, &session_id, "plan", "plan").await;
send_message(&app.app, &session_id, QUESTION_PROMPT).await;
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
find_question_id(events).is_some() || has_event_type(events, "error")
})
.await;
let question_id = find_question_id(&events).expect("question.requested missing");
let answers = find_first_answer(&events).unwrap_or_else(|| vec![vec![]]);
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/questions/{question_id}/reply"),
Some(json!({ "answers": answers })),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "question reply");
let resolved = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
events.iter().any(|event| {
event.get("type").and_then(Value::as_str) == Some("question.resolved")
})
})
.await;
assert!(
resolved.iter().any(|event| {
event.get("type").and_then(Value::as_str) == Some("question.resolved")
&& event
.get("synthetic")
.and_then(Value::as_bool)
.unwrap_or(false)
}),
"question.resolved should be synthetic"
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_agnostic_termination() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
for config in &configs {
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("terminate-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, "default").await;
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/terminate"),
None,
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "terminate session");
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(30), |events| {
has_event_type(events, "session.ended")
})
.await;
assert!(has_event_type(&events, "session.ended"), "missing session.ended");
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/messages"),
Some(json!({ "message": PROMPT })),
)
.await;
assert!(!status.is_success(), "terminated session should reject messages");
}
}

View file

@ -0,0 +1,46 @@
mod common;
use common::*;
use sandbox_agent_agent_management::testing::test_agents_from_env;
use serde_json::Value;
use std::time::Duration;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_basic_reply() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("basic-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, "default").await;
send_message(&app.app, &session_id, PROMPT).await;
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
has_event_type(events, "error") || find_assistant_message_item(events).is_some()
})
.await;
assert!(
!events.is_empty(),
"no events collected for {}",
config.agent.as_str()
);
expect_basic_sequence(&events);
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if caps.tool_calls {
assert!(
!events.iter().any(|event| {
event.get("type").and_then(Value::as_str) == Some("agent.unparsed")
}),
"agent.unparsed event detected"
);
}
}
}

View file

@ -0,0 +1,457 @@
//! Tests for multi-turn conversations to validate session resumption behavior.
//!
//! This test validates that:
//! 1. Sessions can handle multiple messages (multi-turn conversations)
//! 2. Agents that support resumption (Claude, Amp, OpenCode) can continue after process exit
//! 3. Codex supports multi-turn via the shared app-server model (single process, multiple threads)
//! 4. The mock agent correctly supports multi-turn as the reference implementation
use std::time::{Duration, Instant};
use axum::body::Body;
use axum::http::{Method, Request, StatusCode};
use axum::Router;
use http_body_util::BodyExt;
use serde_json::{json, Value};
use tempfile::TempDir;
use sandbox_agent::router::{build_router, AppState, AuthConfig};
use sandbox_agent_agent_management::agents::{AgentId, AgentManager};
use sandbox_agent_agent_management::testing::test_agents_from_env;
use sandbox_agent_agent_credentials::ExtractedCredentials;
use std::collections::BTreeMap;
use tower::util::ServiceExt;
const FIRST_PROMPT: &str = "Reply with exactly the word FIRST.";
const SECOND_PROMPT: &str = "Reply with exactly the word SECOND.";
struct TestApp {
app: Router,
_install_dir: TempDir,
}
impl TestApp {
fn new() -> Self {
let install_dir = tempfile::tempdir().expect("create temp install dir");
let manager = AgentManager::new(install_dir.path()).expect("create agent manager");
let state = AppState::new(AuthConfig::disabled(), manager);
let app = build_router(state);
Self {
app,
_install_dir: install_dir,
}
}
}
struct EnvGuard {
saved: BTreeMap<String, Option<String>>,
}
impl Drop for EnvGuard {
fn drop(&mut self) {
for (key, value) in &self.saved {
match value {
Some(value) => std::env::set_var(key, value),
None => std::env::remove_var(key),
}
}
}
}
fn apply_credentials(creds: &ExtractedCredentials) -> EnvGuard {
let keys = [
"ANTHROPIC_API_KEY",
"CLAUDE_API_KEY",
"OPENAI_API_KEY",
"CODEX_API_KEY",
];
let mut saved = BTreeMap::new();
for key in keys {
saved.insert(key.to_string(), std::env::var(key).ok());
}
match creds.anthropic.as_ref() {
Some(cred) => {
std::env::set_var("ANTHROPIC_API_KEY", &cred.api_key);
std::env::set_var("CLAUDE_API_KEY", &cred.api_key);
}
None => {
std::env::remove_var("ANTHROPIC_API_KEY");
std::env::remove_var("CLAUDE_API_KEY");
}
}
match creds.openai.as_ref() {
Some(cred) => {
std::env::set_var("OPENAI_API_KEY", &cred.api_key);
std::env::set_var("CODEX_API_KEY", &cred.api_key);
}
None => {
std::env::remove_var("OPENAI_API_KEY");
std::env::remove_var("CODEX_API_KEY");
}
}
EnvGuard { saved }
}
async fn send_json(
app: &Router,
method: Method,
path: &str,
body: Option<Value>,
) -> (StatusCode, Value) {
let mut builder = Request::builder().method(method).uri(path);
let body = if let Some(body) = body {
builder = builder.header("content-type", "application/json");
Body::from(body.to_string())
} else {
Body::empty()
};
let request = builder.body(body).expect("request");
let response = app.clone().oneshot(request).await.expect("request handled");
let status = response.status();
let bytes = response
.into_body()
.collect()
.await
.expect("read body")
.to_bytes();
let value = if bytes.is_empty() {
Value::Null
} else {
serde_json::from_slice(&bytes)
.unwrap_or(Value::String(String::from_utf8_lossy(&bytes).to_string()))
};
(status, value)
}
async fn send_status(app: &Router, method: Method, path: &str, body: Option<Value>) -> StatusCode {
let (status, _) = send_json(app, method, path, body).await;
status
}
async fn install_agent(app: &Router, agent: AgentId) {
let status = send_status(
app,
Method::POST,
&format!("/v1/agents/{}/install", agent.as_str()),
Some(json!({})),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "install {agent}");
}
fn test_permission_mode(agent: AgentId) -> &'static str {
match agent {
AgentId::Opencode => "default",
_ => "bypass",
}
}
async fn create_session(app: &Router, agent: AgentId, session_id: &str) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}"),
Some(json!({
"agent": agent.as_str(),
"permissionMode": test_permission_mode(agent)
})),
)
.await;
assert_eq!(status, StatusCode::OK, "create session {agent}");
}
/// Send a message and return the status code (allows checking for errors)
async fn send_message_with_status(
app: &Router,
session_id: &str,
message: &str,
) -> (StatusCode, Value) {
send_json(
app,
Method::POST,
&format!("/v1/sessions/{session_id}/messages"),
Some(json!({ "message": message })),
)
.await
}
/// Wait for a specific number of assistant responses (item.completed with role=assistant)
async fn wait_for_n_responses(
app: &Router,
session_id: &str,
n: usize,
timeout: Duration,
) -> bool {
let start = Instant::now();
while start.elapsed() < timeout {
let path = format!("/v1/sessions/{session_id}/events?offset=0&limit=1000");
let (status, payload) = send_json(app, Method::GET, &path, None).await;
if status != StatusCode::OK {
return false;
}
let events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let completed_count = events.iter().filter(|e| is_assistant_completed(e)).count();
if completed_count >= n {
return true;
}
// Check for errors
for event in &events {
if is_error_event(event) {
eprintln!("Error event: {:?}", event);
return false;
}
}
tokio::time::sleep(Duration::from_millis(300)).await;
}
false
}
/// Wait for an assistant response (item.completed with role=assistant)
async fn wait_for_response(app: &Router, session_id: &str, timeout: Duration) -> bool {
wait_for_n_responses(app, session_id, 1, timeout).await
}
fn is_assistant_completed(event: &Value) -> bool {
event
.get("type")
.and_then(Value::as_str)
.map(|t| t == "item.completed")
.unwrap_or(false)
&& event
.get("data")
.and_then(|d| d.get("item"))
.and_then(|i| i.get("role"))
.and_then(Value::as_str)
.map(|r| r == "assistant")
.unwrap_or(false)
}
fn is_session_ended(event: &Value) -> bool {
event
.get("type")
.and_then(Value::as_str)
.map(|t| t == "session.ended")
.unwrap_or(false)
}
fn is_error_event(event: &Value) -> bool {
matches!(
event.get("type").and_then(Value::as_str),
Some("error") | Some("agent.unparsed")
)
}
/// Count assistant responses in the event stream
async fn count_assistant_responses(app: &Router, session_id: &str) -> usize {
let path = format!("/v1/sessions/{session_id}/events?offset=0&limit=1000");
let (status, payload) = send_json(app, Method::GET, &path, None).await;
if status != StatusCode::OK {
eprintln!("Failed to get events: status={}", status);
return 0;
}
let events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
// Debug: print all event types
eprintln!("All events ({}):", events.len());
for (i, e) in events.iter().enumerate() {
let event_type = e.get("type").and_then(Value::as_str).unwrap_or("?");
let role = e
.get("data")
.and_then(|d| d.get("item"))
.and_then(|i| i.get("role"))
.and_then(Value::as_str)
.unwrap_or("-");
eprintln!(" [{}] type={}, role={}", i, event_type, role);
}
let count = events.iter().filter(|e| is_assistant_completed(e)).count();
eprintln!("Assistant completed count: {}", count);
count
}
/// Test multi-turn conversation for a specific agent
async fn test_multi_turn_for_agent(app: &Router, agent: AgentId) -> Result<(), String> {
let session_id = format!("multi-turn-{}", agent.as_str());
eprintln!("\n=== Testing multi-turn for {} ===", agent);
// Create session
create_session(app, agent, &session_id).await;
eprintln!("Session created: {}", session_id);
// Send first message
eprintln!("Sending first message...");
let (status, body) = send_message_with_status(app, &session_id, FIRST_PROMPT).await;
eprintln!("First message status: {}", status);
if status != StatusCode::NO_CONTENT {
return Err(format!(
"First message failed with status {}: {:?}",
status, body
));
}
// Wait for first response
eprintln!("Waiting for first response...");
let got_first = wait_for_response(app, &session_id, Duration::from_secs(120)).await;
if !got_first {
return Err("Timed out waiting for first response".to_string());
}
eprintln!("Got first response");
// Small delay to ensure session state is updated
tokio::time::sleep(Duration::from_millis(500)).await;
// Send second message - this is the critical test
eprintln!("Sending second message...");
let (status, body) = send_message_with_status(app, &session_id, SECOND_PROMPT).await;
eprintln!("Second message status: {}, body: {:?}", status, body);
if status != StatusCode::NO_CONTENT {
return Err(format!(
"Second message failed with status {}: {:?}",
status, body
));
}
// Wait for second response - specifically wait for 2 completed responses
eprintln!("Waiting for second response (total 2)...");
let got_both = wait_for_n_responses(app, &session_id, 2, Duration::from_secs(120)).await;
if !got_both {
// Debug: show what we got
let response_count = count_assistant_responses(app, &session_id).await;
return Err(format!(
"Timed out waiting for second response (got {} completed)",
response_count
));
}
eprintln!("Got both responses");
// Verify we got two assistant responses
let response_count = count_assistant_responses(app, &session_id).await;
eprintln!("Final response count: {}", response_count);
if response_count < 2 {
return Err(format!(
"Expected at least 2 assistant responses, got {}",
response_count
));
}
Ok(())
}
#[tokio::test]
async fn multi_turn_mock_agent() {
let test_app = TestApp::new();
// Mock agent should always support multi-turn as the reference implementation
let result = test_multi_turn_for_agent(&test_app.app, AgentId::Mock).await;
assert!(
result.is_ok(),
"Mock agent multi-turn failed: {:?}",
result.err()
);
}
#[tokio::test]
async fn multi_turn_real_agents() {
let configs = match test_agents_from_env() {
Ok(configs) => configs,
Err(err) => {
eprintln!("Failed to get agent configs: {:?}. Skipping multi-turn test.", err);
return;
}
};
if configs.is_empty() {
eprintln!("No agents configured for testing. Skipping multi-turn test.");
return;
}
let test_app = TestApp::new();
for config in configs {
let _guard = apply_credentials(&config.credentials);
install_agent(&test_app.app, config.agent).await;
let result = test_multi_turn_for_agent(&test_app.app, config.agent).await;
match config.agent {
AgentId::Claude | AgentId::Amp | AgentId::Opencode => {
// These agents should support multi-turn via resumption
assert!(
result.is_ok(),
"{} multi-turn failed (should support resumption): {:?}",
config.agent,
result.err()
);
}
AgentId::Codex => {
// Codex now supports multi-turn via the shared app-server model
assert!(
result.is_ok(),
"{} multi-turn failed (should support shared app-server): {:?}",
config.agent,
result.err()
);
}
AgentId::Mock => {
// Mock is tested separately
}
}
}
}
/// Test that verifies the session can be reopened after ending
#[tokio::test]
async fn session_reopen_after_end() {
let test_app = TestApp::new();
let session_id = "reopen-test";
// Create session with mock agent
create_session(&test_app.app, AgentId::Mock, session_id).await;
// Send "end" command to mock agent to end the session
let (status, _) = send_message_with_status(&test_app.app, session_id, "end").await;
assert_eq!(status, StatusCode::NO_CONTENT);
// Wait for session to end
tokio::time::sleep(Duration::from_millis(500)).await;
// Verify session is ended
let path = format!("/v1/sessions/{session_id}/events?offset=0&limit=100");
let (_, payload) = send_json(&test_app.app, Method::GET, &path, None).await;
let events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let has_ended = events.iter().any(|e| is_session_ended(e));
assert!(has_ended, "Session should be ended after 'end' command");
// Try to send another message - mock agent supports resume so this should work
// (or fail if we haven't implemented reopen for mock)
let (status, body) = send_message_with_status(&test_app.app, session_id, "hello again").await;
// For mock agent, the session should be reopenable since mock is in agent_supports_resume
// But mock's session.ended is triggered differently than real agents
// This test documents the current behavior
if status == StatusCode::NO_CONTENT {
eprintln!("Mock agent session was successfully reopened after end");
} else {
eprintln!(
"Mock agent session could not be reopened (status {}): {:?}",
status, body
);
}
}

View file

@ -0,0 +1,63 @@
mod common;
use common::*;
use sandbox_agent_agent_management::testing::test_agents_from_env;
use std::time::Duration;
use axum::http::Method;
use serde_json::json;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_permission_flow() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if !(caps.plan_mode && caps.permissions) {
continue;
}
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("perm-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, "plan").await;
send_message(&app.app, &session_id, TOOL_PROMPT).await;
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
find_permission_id(events).is_some() || has_event_type(events, "error")
})
.await;
let permission_id = find_permission_id(&events).expect("permission.requested missing");
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/permissions/{permission_id}/reply"),
Some(json!({ "reply": "once" })),
)
.await;
assert_eq!(status, axum::http::StatusCode::NO_CONTENT, "permission reply");
let resolved = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
events.iter().any(|event| {
event.get("type").and_then(serde_json::Value::as_str) == Some("permission.resolved")
})
})
.await;
assert!(
resolved.iter().any(|event| {
event.get("type").and_then(serde_json::Value::as_str) == Some("permission.resolved")
&& event
.get("synthetic")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
}),
"permission.resolved should be synthetic"
);
}
}

View file

@ -0,0 +1,64 @@
mod common;
use common::*;
use sandbox_agent_agent_management::testing::test_agents_from_env;
use std::time::Duration;
use axum::http::Method;
use serde_json::json;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_question_flow() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if !caps.questions {
continue;
}
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("question-{}", config.agent.as_str());
create_session_with_mode(&app.app, config.agent, &session_id, "plan", "plan").await;
send_message(&app.app, &session_id, QUESTION_PROMPT).await;
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
find_question_id(events).is_some() || has_event_type(events, "error")
})
.await;
let question_id = find_question_id(&events).expect("question.requested missing");
let answers = find_first_answer(&events).unwrap_or_else(|| vec![vec![]]);
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/questions/{question_id}/reply"),
Some(json!({ "answers": answers })),
)
.await;
assert_eq!(status, axum::http::StatusCode::NO_CONTENT, "question reply");
let resolved = poll_events_until(&app.app, &session_id, Duration::from_secs(120), |events| {
events.iter().any(|event| {
event.get("type").and_then(serde_json::Value::as_str) == Some("question.resolved")
})
})
.await;
assert!(
resolved.iter().any(|event| {
event.get("type").and_then(serde_json::Value::as_str) == Some("question.resolved")
&& event
.get("synthetic")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
}),
"question.resolved should be synthetic"
);
}
}

View file

@ -0,0 +1,45 @@
mod common;
use common::*;
use sandbox_agent_agent_management::testing::test_agents_from_env;
use std::time::Duration;
use axum::http::Method;
use serde_json::json;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_termination() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
for config in &configs {
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("terminate-{}", config.agent.as_str());
create_session(&app.app, config.agent, &session_id, "default").await;
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/terminate"),
None,
)
.await;
assert_eq!(status, axum::http::StatusCode::NO_CONTENT, "terminate session");
let events = poll_events_until(&app.app, &session_id, Duration::from_secs(30), |events| {
has_event_type(events, "session.ended")
})
.await;
assert!(has_event_type(&events, "session.ended"), "missing session.ended");
let status = send_status(
&app.app,
Method::POST,
&format!("/v1/sessions/{session_id}/messages"),
Some(json!({ "message": PROMPT })),
)
.await;
assert!(!status.is_success(), "terminated session should reject messages");
}
}

View file

@ -0,0 +1,94 @@
mod common;
use common::*;
use sandbox_agent_agent_management::testing::test_agents_from_env;
use serde_json::Value;
use std::time::{Duration, Instant};
use axum::http::Method;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_tool_flow() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
let capabilities = fetch_capabilities(&app.app).await;
for config in &configs {
let caps = capabilities
.get(config.agent.as_str())
.expect("capabilities missing");
if !caps.tool_calls {
continue;
}
let _guard = apply_credentials(&config.credentials);
install_agent(&app.app, config.agent).await;
let session_id = format!("tool-{}", config.agent.as_str());
create_session(
&app.app,
config.agent,
&session_id,
test_permission_mode(config.agent),
)
.await;
send_message(&app.app, &session_id, TOOL_PROMPT).await;
let start = Instant::now();
let mut offset = 0u64;
let mut events = Vec::new();
let mut replied = false;
while start.elapsed() < Duration::from_secs(180) {
let path = format!("/v1/sessions/{session_id}/events?offset={offset}&limit=200");
let (status, payload) = send_json(&app.app, Method::GET, &path, None).await;
assert_eq!(status, axum::http::StatusCode::OK, "poll events");
let new_events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if !new_events.is_empty() {
if let Some(last) = new_events
.last()
.and_then(|event| event.get("sequence"))
.and_then(Value::as_u64)
{
offset = last;
}
events.extend(new_events);
if !replied {
if let Some(permission_id) = find_permission_id(&events) {
let _ = send_status(
&app.app,
Method::POST,
&format!(
"/v1/sessions/{session_id}/permissions/{permission_id}/reply"
),
Some(serde_json::json!({ "reply": "once" })),
)
.await;
replied = true;
}
}
if has_tool_result(&events) {
break;
}
}
tokio::time::sleep(Duration::from_millis(800)).await;
}
let tool_call = find_tool_call(&events);
let tool_result = has_tool_result(&events);
assert!(
tool_call.is_some(),
"tool_call missing for tool-capable agent {}",
config.agent.as_str()
);
if tool_call.is_some() {
assert!(
tool_result,
"tool_result missing after tool_call for {}",
config.agent.as_str()
);
}
}
}

View file

@ -0,0 +1,388 @@
use std::collections::HashMap;
use std::time::{Duration, Instant};
use axum::body::Body;
use axum::http::{Method, Request, StatusCode};
use axum::Router;
use http_body_util::BodyExt;
use serde_json::{json, Value};
use tempfile::TempDir;
use tower::util::ServiceExt;
use sandbox_agent::router::{
build_router,
AgentCapabilities,
AgentListResponse,
AuthConfig,
};
use sandbox_agent_agent_credentials::ExtractedCredentials;
use sandbox_agent_agent_management::agents::{AgentId, AgentManager};
pub const PROMPT: &str = "Reply with exactly the single word OK.";
pub const TOOL_PROMPT: &str =
"Use the bash tool to run `ls` in the current directory. Do not answer without using the tool.";
pub const QUESTION_PROMPT: &str =
"Call the AskUserQuestion tool with exactly one yes/no question and wait for a reply. Do not answer yourself.";
pub struct TestApp {
pub app: Router,
_install_dir: TempDir,
}
impl TestApp {
pub fn new() -> Self {
let install_dir = tempfile::tempdir().expect("create temp install dir");
let manager = AgentManager::new(install_dir.path())
.expect("create agent manager");
let state = sandbox_agent::router::AppState::new(AuthConfig::disabled(), manager);
let app = build_router(state);
Self {
app,
_install_dir: install_dir,
}
}
}
pub struct EnvGuard {
saved: HashMap<String, Option<String>>,
}
impl Drop for EnvGuard {
fn drop(&mut self) {
for (key, value) in &self.saved {
match value {
Some(value) => std::env::set_var(key, value),
None => std::env::remove_var(key),
}
}
}
}
pub fn apply_credentials(creds: &ExtractedCredentials) -> EnvGuard {
let keys = ["ANTHROPIC_API_KEY", "CLAUDE_API_KEY", "OPENAI_API_KEY", "CODEX_API_KEY"];
let mut saved = HashMap::new();
for key in keys {
saved.insert(key.to_string(), std::env::var(key).ok());
}
match creds.anthropic.as_ref() {
Some(cred) => {
std::env::set_var("ANTHROPIC_API_KEY", &cred.api_key);
std::env::set_var("CLAUDE_API_KEY", &cred.api_key);
}
None => {
std::env::remove_var("ANTHROPIC_API_KEY");
std::env::remove_var("CLAUDE_API_KEY");
}
}
match creds.openai.as_ref() {
Some(cred) => {
std::env::set_var("OPENAI_API_KEY", &cred.api_key);
std::env::set_var("CODEX_API_KEY", &cred.api_key);
}
None => {
std::env::remove_var("OPENAI_API_KEY");
std::env::remove_var("CODEX_API_KEY");
}
}
EnvGuard { saved }
}
pub async fn send_json(
app: &Router,
method: Method,
path: &str,
body: Option<Value>,
) -> (StatusCode, Value) {
let request = Request::builder()
.method(method)
.uri(path)
.header("content-type", "application/json")
.body(Body::from(body.map(|value| value.to_string()).unwrap_or_default()))
.expect("request");
let response = app
.clone()
.oneshot(request)
.await
.expect("response");
let status = response.status();
let bytes = response
.into_body()
.collect()
.await
.expect("body")
.to_bytes();
let payload = if bytes.is_empty() {
Value::Null
} else {
serde_json::from_slice(&bytes).unwrap_or(Value::Null)
};
(status, payload)
}
pub async fn send_status(
app: &Router,
method: Method,
path: &str,
body: Option<Value>,
) -> StatusCode {
let (status, _) = send_json(app, method, path, body).await;
status
}
pub async fn install_agent(app: &Router, agent: AgentId) {
let status = send_status(
app,
Method::POST,
&format!("/v1/agents/{}/install", agent.as_str()),
Some(json!({})),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "install agent {}", agent.as_str());
}
pub async fn create_session(
app: &Router,
agent: AgentId,
session_id: &str,
permission_mode: &str,
) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}"),
Some(json!({
"agent": agent.as_str(),
"permissionMode": permission_mode,
})),
)
.await;
assert_eq!(status, StatusCode::OK, "create session");
}
pub async fn create_session_with_mode(
app: &Router,
agent: AgentId,
session_id: &str,
agent_mode: &str,
permission_mode: &str,
) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}"),
Some(json!({
"agent": agent.as_str(),
"agentMode": agent_mode,
"permissionMode": permission_mode,
})),
)
.await;
assert_eq!(status, StatusCode::OK, "create session");
}
pub fn test_permission_mode(agent: AgentId) -> &'static str {
match agent {
AgentId::Opencode => "default",
_ => "bypass",
}
}
pub async fn send_message(app: &Router, session_id: &str, message: &str) {
let status = send_status(
app,
Method::POST,
&format!("/v1/sessions/{session_id}/messages"),
Some(json!({ "message": message })),
)
.await;
assert_eq!(status, StatusCode::NO_CONTENT, "send message");
}
pub async fn poll_events_until<F>(
app: &Router,
session_id: &str,
timeout: Duration,
mut stop: F,
) -> Vec<Value>
where
F: FnMut(&[Value]) -> bool,
{
let start = Instant::now();
let mut offset = 0u64;
let mut events = Vec::new();
while start.elapsed() < timeout {
let path = format!("/v1/sessions/{session_id}/events?offset={offset}&limit=200");
let (status, payload) = send_json(app, Method::GET, &path, None).await;
assert_eq!(status, StatusCode::OK, "poll events");
let new_events = payload
.get("events")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if !new_events.is_empty() {
if let Some(last) = new_events
.last()
.and_then(|event| event.get("sequence"))
.and_then(Value::as_u64)
{
offset = last;
}
events.extend(new_events);
if stop(&events) {
break;
}
}
tokio::time::sleep(Duration::from_millis(800)).await;
}
events
}
pub async fn fetch_capabilities(app: &Router) -> HashMap<String, AgentCapabilities> {
let (status, payload) = send_json(app, Method::GET, "/v1/agents", None).await;
assert_eq!(status, StatusCode::OK, "list agents");
let response: AgentListResponse = serde_json::from_value(payload).expect("agents payload");
response
.agents
.into_iter()
.map(|agent| (agent.id, agent.capabilities))
.collect()
}
pub fn has_event_type(events: &[Value], event_type: &str) -> bool {
events
.iter()
.any(|event| event.get("type").and_then(Value::as_str) == Some(event_type))
}
pub fn find_assistant_message_item(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("item.completed") {
return None;
}
let item = event.get("data")?.get("item")?;
let role = item.get("role")?.as_str()?;
let kind = item.get("kind")?.as_str()?;
if role != "assistant" || kind != "message" {
return None;
}
item.get("item_id")?.as_str().map(|id| id.to_string())
})
}
pub fn event_sequence(event: &Value) -> Option<u64> {
event.get("sequence").and_then(Value::as_u64)
}
pub fn find_item_event_seq(events: &[Value], event_type: &str, item_id: &str) -> Option<u64> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some(event_type) {
return None;
}
match event_type {
"item.delta" => {
let data = event.get("data")?;
let id = data.get("item_id")?.as_str()?;
if id == item_id {
event_sequence(event)
} else {
None
}
}
_ => {
let item = event.get("data")?.get("item")?;
let id = item.get("item_id")?.as_str()?;
if id == item_id {
event_sequence(event)
} else {
None
}
}
}
})
}
pub fn find_permission_id(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("permission.requested") {
return None;
}
event
.get("data")
.and_then(|data| data.get("permission_id"))
.and_then(Value::as_str)
.map(|id| id.to_string())
})
}
pub fn find_question_id(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("question.requested") {
return None;
}
event
.get("data")
.and_then(|data| data.get("question_id"))
.and_then(Value::as_str)
.map(|id| id.to_string())
})
}
pub fn find_first_answer(events: &[Value]) -> Option<Vec<Vec<String>>> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("question.requested") {
return None;
}
let options = event
.get("data")
.and_then(|data| data.get("options"))
.and_then(Value::as_array)?;
let option = options.first()?.as_str()?.to_string();
Some(vec![vec![option]])
})
}
pub fn find_tool_call(events: &[Value]) -> Option<String> {
events.iter().find_map(|event| {
if event.get("type").and_then(Value::as_str) != Some("item.started")
&& event.get("type").and_then(Value::as_str) != Some("item.completed")
{
return None;
}
let item = event.get("data")?.get("item")?;
let kind = item.get("kind")?.as_str()?;
if kind != "tool_call" {
return None;
}
item.get("item_id")?.as_str().map(|id| id.to_string())
})
}
pub fn has_tool_result(events: &[Value]) -> bool {
events.iter().any(|event| {
if event.get("type").and_then(Value::as_str) != Some("item.completed") {
return false;
}
let item = match event.get("data").and_then(|data| data.get("item")) {
Some(item) => item,
None => return false,
};
item.get("kind").and_then(Value::as_str) == Some("tool_result")
})
}
pub fn expect_basic_sequence(events: &[Value]) {
assert!(has_event_type(events, "session.started"), "session.started missing");
let item_id = find_assistant_message_item(events).expect("assistant message missing");
let started_seq = find_item_event_seq(events, "item.started", &item_id)
.expect("item.started missing");
// Intentionally require deltas here to validate our synthetic delta behavior.
let delta_seq = find_item_event_seq(events, "item.delta", &item_id)
.expect("item.delta missing");
let completed_seq = find_item_event_seq(events, "item.completed", &item_id)
.expect("item.completed missing");
assert!(started_seq < delta_seq, "item.started must precede delta");
assert!(delta_seq < completed_seq, "delta must precede completion");
}

View file

@ -291,6 +291,57 @@ async fn read_sse_events(
events
}
async fn read_turn_stream_events(
app: &Router,
session_id: &str,
timeout: Duration,
) -> Vec<Value> {
let request = Request::builder()
.method(Method::POST)
.uri(format!("/v1/sessions/{session_id}/messages/stream"))
.header("content-type", "application/json")
.body(Body::from(json!({ "message": PROMPT }).to_string()))
.expect("turn stream request");
let response = app
.clone()
.oneshot(request)
.await
.expect("turn stream response");
assert_eq!(response.status(), StatusCode::OK, "turn stream status");
let mut stream = response.into_body().into_data_stream();
let mut buffer = String::new();
let mut events = Vec::new();
let start = Instant::now();
let mut ended = false;
loop {
let remaining = match timeout.checked_sub(start.elapsed()) {
Some(remaining) if !remaining.is_zero() => remaining,
_ => break,
};
let next = tokio::time::timeout(remaining, stream.next()).await;
let chunk: Bytes = match next {
Ok(Some(Ok(chunk))) => chunk,
Ok(Some(Err(_))) => break,
Ok(None) => {
ended = true;
break;
}
Err(_) => break,
};
buffer.push_str(&String::from_utf8_lossy(&chunk));
while let Some(idx) = buffer.find("\n\n") {
let block = buffer[..idx].to_string();
buffer = buffer[idx + 2..].to_string();
if let Some(event) = parse_sse_block(&block) {
events.push(event);
}
}
}
assert!(ended, "turn stream did not close before timeout");
events
}
fn parse_sse_block(block: &str) -> Option<Value> {
let mut data_lines = Vec::new();
for line in block.lines() {
@ -798,6 +849,27 @@ async fn run_sse_events_snapshot(app: &Router, config: &TestAgentConfig) {
});
}
async fn run_turn_stream_check(app: &Router, config: &TestAgentConfig) {
let _guard = apply_credentials(&config.credentials);
install_agent(app, config.agent).await;
let session_id = format!("turn-{}", config.agent.as_str());
create_session(app, config.agent, &session_id, test_permission_mode(config.agent)).await;
let events = read_turn_stream_events(app, &session_id, Duration::from_secs(120)).await;
let events = truncate_after_first_stop(&events);
assert!(
!events.is_empty(),
"no turn stream events collected for {}",
config.agent
);
assert!(
should_stop(&events),
"timed out waiting for assistant/error event for {}",
config.agent
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn auth_snapshots() {
let token = "test-token";
@ -1294,6 +1366,20 @@ async fn sse_events_snapshots() {
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn turn_stream_route() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");
let app = TestApp::new();
for config in &configs {
// OpenCode's embedded bun hangs when installing plugins, blocking SSE event streaming.
// See: https://github.com/opencode-ai/opencode/issues/XXX
if config.agent == AgentId::Opencode {
continue;
}
run_turn_stream_check(&app.app, config).await;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn concurrency_snapshots() {
let configs = test_agents_from_env().expect("configure SANDBOX_TEST_AGENTS or install agents");

View file

@ -1,10 +1,10 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 881
expression: normalize_agent_list(&agents)
---
agents:
- id: amp
- id: claude
- id: codex
- id: mock
- id: opencode

View file

@ -1,6 +1,6 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 934
expression: normalize_create_session(&created)
---
healthy: true
nativeSessionId: "<redacted>"

View file

@ -1,36 +1,379 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 984
expression: normalize_events(&permission_events)
---
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- metadata: true
seq: 2
started:
message: thread/started
- agent: codex
kind: started
session: started
source: agent
synthetic: false
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 5
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 6
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 7
source: agent
synthetic: false
type: item.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 8
source: agent
synthetic: false
type: item.completed
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 9
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 10
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 11
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 12
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 13
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 14
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 15
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 16
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 17
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 18
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 19
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 20
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 21
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 22
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 23
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 24
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 25
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 26
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 27
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 28
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 29
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 30
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 31
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 32
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 33
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 34
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 35
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 36
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 37
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 38
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 39
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 40
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 41
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 42
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 43
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 44
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 45
source: agent
synthetic: false
type: item.delta
- item:
content_types:
- reasoning
kind: message
role: assistant
status: completed
seq: 46
source: agent
synthetic: false
type: item.completed

View file

@ -1,36 +1,275 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 1106
expression: normalize_events(&reject_events)
---
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- metadata: true
seq: 2
started:
message: thread/started
- agent: codex
kind: started
session: started
source: agent
synthetic: false
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 5
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 6
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 7
source: agent
synthetic: false
type: item.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 8
source: agent
synthetic: false
type: item.completed
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 9
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 10
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 11
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 12
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 13
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 14
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 15
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 16
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 17
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 18
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 19
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 20
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 21
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 22
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 23
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 24
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 25
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 26
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 27
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 28
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 29
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 30
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 31
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 32
source: agent
synthetic: false
type: item.delta
- item:
content_types:
- reasoning
kind: message
role: assistant
status: completed
seq: 33
source: agent
synthetic: false
type: item.completed

View file

@ -1,36 +1,72 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 1045
expression: normalize_events(&question_events)
---
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- metadata: true
seq: 2
started:
message: thread/started
- agent: codex
kind: started
session: started
source: agent
synthetic: false
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 5
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 6
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 7
source: agent
synthetic: false
type: item.started
- item:
content_types: []
kind: message
role: assistant
status: completed
seq: 8
source: agent
synthetic: false
type: item.completed

View file

@ -1,6 +1,5 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 810
expression: "json!({\n \"status\": status.as_u16(), \"payload\": normalize_agent_list(&payload),\n})"
---
payload:
@ -8,5 +7,6 @@ payload:
- id: amp
- id: claude
- id: codex
- id: mock
- id: opencode
status: 200

View file

@ -1,69 +1,224 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 1214
expression: snapshot
---
session_a:
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 2
started:
message: thread/started
- agent: codex
kind: started
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 3
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 5
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 6
source: agent
synthetic: false
type: item.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 7
source: agent
synthetic: false
type: item.completed
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 8
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 9
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 10
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 11
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 12
source: agent
synthetic: false
type: item.delta
- item:
content_types:
- reasoning
kind: message
role: assistant
status: completed
seq: 13
source: agent
synthetic: false
type: item.completed
session_b:
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 2
started:
message: thread/started
- agent: codex
kind: started
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 3
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 5
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 6
source: agent
synthetic: false
type: item.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 7
source: agent
synthetic: false
type: item.completed
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 8
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 9
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 10
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 11
source: agent
synthetic: false
type: item.delta
- item:
content_types:
- reasoning
kind: message
role: assistant
status: completed
seq: 12
source: agent
synthetic: false
type: item.completed

View file

@ -1,36 +1,91 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 697
expression: normalized
---
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- metadata: true
seq: 2
started:
message: thread/started
- agent: codex
kind: started
session: started
source: agent
synthetic: false
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 5
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 6
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 7
source: agent
synthetic: false
type: item.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 8
source: agent
synthetic: false
type: item.completed
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 9
source: agent
synthetic: false
type: item.delta
- item:
content_types:
- reasoning
kind: message
role: assistant
status: completed
seq: 10
source: agent
synthetic: false
type: item.completed

View file

@ -1,36 +1,109 @@
---
source: server/packages/sandbox-agent/tests/http_sse_snapshots.rs
assertion_line: 734
expression: normalized
---
- agent: codex
kind: started
- metadata: true
seq: 1
started:
message: session.created
- agent: codex
kind: started
session: started
source: daemon
synthetic: true
type: session.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 2
started:
message: thread/started
- agent: codex
kind: started
seq: 3
started:
message: turn/started
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
source: agent
synthetic: false
type: item.completed
- item:
content_types:
- text
kind: message
role: user
status: in_progress
seq: 3
source: agent
synthetic: false
type: item.started
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 4
- agent: codex
kind: message
message:
parts:
- text: "<redacted>"
type: text
role: assistant
source: daemon
synthetic: true
type: item.delta
- item:
content_types:
- text
kind: message
role: user
status: completed
seq: 5
source: agent
synthetic: false
type: item.completed
- item:
content_types: []
kind: message
role: assistant
status: in_progress
seq: 6
source: agent
synthetic: false
type: item.started
- item:
content_types:
- status
kind: status
role: system
status: completed
seq: 7
source: agent
synthetic: false
type: item.completed
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 8
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 9
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 10
source: agent
synthetic: false
type: item.delta
- delta:
delta: "<redacted>"
item_id: "<redacted>"
native_item_id: "<redacted>"
seq: 11
source: agent
synthetic: false
type: item.delta
- item:
content_types:
- reasoning
kind: message
role: assistant
status: completed
seq: 12
source: agent
synthetic: false
type: item.completed

View file

@ -31,10 +31,10 @@ pub fn event_to_universal_with_session(
let event_type = event.get("type").and_then(Value::as_str).unwrap_or("");
let mut conversions = match event_type {
"system" => vec![system_event_to_universal(event)],
"assistant" => assistant_event_to_universal(event),
"tool_use" => tool_use_event_to_universal(event, session_id),
"assistant" => assistant_event_to_universal(event, &session_id),
"tool_use" => tool_use_event_to_universal(event, &session_id),
"tool_result" => tool_result_event_to_universal(event),
"result" => result_event_to_universal(event),
"result" => result_event_to_universal(event, &session_id),
_ => return Err(format!("unsupported Claude event type: {event_type}")),
};
@ -53,7 +53,7 @@ fn system_event_to_universal(event: &Value) -> EventConversion {
.with_raw(Some(event.clone()))
}
fn assistant_event_to_universal(event: &Value) -> Vec<EventConversion> {
fn assistant_event_to_universal(event: &Value, session_id: &str) -> Vec<EventConversion> {
let mut conversions = Vec::new();
let content = event
.get("message")
@ -62,7 +62,8 @@ fn assistant_event_to_universal(event: &Value) -> Vec<EventConversion> {
.cloned()
.unwrap_or_default();
let message_id = next_temp_id("tmp_claude_message");
// Use session-based native_item_id so `result` event can reference the same item
let native_message_id = format!("{session_id}_message");
let mut message_parts = Vec::new();
for block in content {
@ -85,9 +86,9 @@ fn assistant_event_to_universal(event: &Value) -> Vec<EventConversion> {
.unwrap_or_else(|| next_temp_id("tmp_claude_tool"));
let arguments = serde_json::to_string(&input).unwrap_or_else(|_| "{}".to_string());
let tool_item = UniversalItem {
item_id: next_temp_id("tmp_claude_tool_item"),
item_id: String::new(),
native_item_id: Some(call_id.clone()),
parent_id: Some(message_id.clone()),
parent_id: Some(native_message_id.clone()),
kind: ItemKind::ToolCall,
role: Some(ItemRole::Assistant),
content: vec![ContentPart::ToolCall {
@ -106,21 +107,23 @@ fn assistant_event_to_universal(event: &Value) -> Vec<EventConversion> {
}
}
// `assistant` event emits item.started + item.delta only (in-progress state)
// The `result` event will emit item.completed to finalize
let message_item = UniversalItem {
item_id: message_id,
native_item_id: None,
item_id: String::new(),
native_item_id: Some(native_message_id.clone()),
parent_id: None,
kind: ItemKind::Message,
role: Some(ItemRole::Assistant),
content: message_parts.clone(),
status: ItemStatus::Completed,
status: ItemStatus::InProgress,
};
conversions.extend(message_events(message_item, message_parts, true));
conversions.extend(message_started_events(message_item, message_parts));
conversions
}
fn tool_use_event_to_universal(event: &Value, session_id: String) -> Vec<EventConversion> {
fn tool_use_event_to_universal(event: &Value, session_id: &str) -> Vec<EventConversion> {
let mut conversions = Vec::new();
let tool_use = event.get("tool_use");
let name = tool_use
@ -156,7 +159,7 @@ fn tool_use_event_to_universal(event: &Value, session_id: String) -> Vec<EventCo
let arguments = serde_json::to_string(&input).unwrap_or_else(|_| "{}".to_string());
let tool_item = UniversalItem {
item_id: next_temp_id("tmp_claude_tool_item"),
item_id: String::new(),
native_item_id: Some(id.clone()),
parent_id: None,
kind: ItemKind::ToolCall,
@ -222,22 +225,30 @@ fn tool_result_event_to_universal(event: &Value) -> Vec<EventConversion> {
conversions
}
fn result_event_to_universal(event: &Value) -> Vec<EventConversion> {
fn result_event_to_universal(event: &Value, session_id: &str) -> Vec<EventConversion> {
// The `result` event completes the message started by `assistant`.
// Use the same native_item_id so they link to the same universal item.
let native_message_id = format!("{session_id}_message");
let result_text = event
.get("result")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let message_item = UniversalItem {
item_id: next_temp_id("tmp_claude_result"),
native_item_id: None,
item_id: String::new(),
native_item_id: Some(native_message_id),
parent_id: None,
kind: ItemKind::Message,
role: Some(ItemRole::Assistant),
content: vec![ContentPart::Text { text: result_text.clone() }],
content: vec![ContentPart::Text { text: result_text }],
status: ItemStatus::Completed,
};
message_events(message_item, vec![ContentPart::Text { text: result_text }], true)
vec![EventConversion::new(
UniversalEventType::ItemCompleted,
UniversalEventData::Item(ItemEventData { item: message_item }),
)]
}
fn item_events(item: UniversalItem, synthetic_start: bool) -> Vec<EventConversion> {
@ -260,20 +271,18 @@ fn item_events(item: UniversalItem, synthetic_start: bool) -> Vec<EventConversio
events
}
fn message_events(item: UniversalItem, parts: Vec<ContentPart>, synthetic_start: bool) -> Vec<EventConversion> {
/// Emits item.started + item.delta only (for `assistant` event).
/// The item.completed will come from the `result` event.
fn message_started_events(item: UniversalItem, parts: Vec<ContentPart>) -> Vec<EventConversion> {
let mut events = Vec::new();
if synthetic_start {
let mut started_item = item.clone();
started_item.status = ItemStatus::InProgress;
events.push(
EventConversion::new(
UniversalEventType::ItemStarted,
UniversalEventData::Item(ItemEventData { item: started_item }),
)
.synthetic(),
);
}
// Emit item.started (in-progress)
events.push(EventConversion::new(
UniversalEventType::ItemStarted,
UniversalEventData::Item(ItemEventData { item: item.clone() }),
));
// Emit item.delta with the text content
let mut delta_text = String::new();
for part in &parts {
if let ContentPart::Text { text } = part {
@ -281,23 +290,16 @@ fn message_events(item: UniversalItem, parts: Vec<ContentPart>, synthetic_start:
}
}
if !delta_text.is_empty() {
events.push(
EventConversion::new(
UniversalEventType::ItemDelta,
UniversalEventData::ItemDelta(crate::ItemDeltaData {
item_id: item.item_id.clone(),
native_item_id: item.native_item_id.clone(),
delta: delta_text,
}),
)
.synthetic(),
);
events.push(EventConversion::new(
UniversalEventType::ItemDelta,
UniversalEventData::ItemDelta(crate::ItemDeltaData {
item_id: item.item_id.clone(),
native_item_id: item.native_item_id.clone(),
delta: delta_text,
}),
));
}
events.push(EventConversion::new(
UniversalEventType::ItemCompleted,
UniversalEventData::Item(ItemEventData { item }),
));
events
}

553
spec/universal-schema.json Normal file
View file

@ -0,0 +1,553 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "UniversalEvent",
"type": "object",
"required": [
"data",
"event_id",
"sequence",
"session_id",
"source",
"synthetic",
"time",
"type"
],
"properties": {
"data": {
"$ref": "#/definitions/UniversalEventData"
},
"event_id": {
"type": "string"
},
"native_session_id": {
"type": [
"string",
"null"
]
},
"raw": true,
"sequence": {
"type": "integer",
"format": "uint64",
"minimum": 0.0
},
"session_id": {
"type": "string"
},
"source": {
"$ref": "#/definitions/EventSource"
},
"synthetic": {
"type": "boolean"
},
"time": {
"type": "string"
},
"type": {
"$ref": "#/definitions/UniversalEventType"
}
},
"definitions": {
"AgentUnparsedData": {
"type": "object",
"required": [
"error",
"location"
],
"properties": {
"error": {
"type": "string"
},
"location": {
"type": "string"
},
"raw_hash": {
"type": [
"string",
"null"
]
}
}
},
"ContentPart": {
"oneOf": [
{
"type": "object",
"required": [
"text",
"type"
],
"properties": {
"text": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"text"
]
}
}
},
{
"type": "object",
"required": [
"json",
"type"
],
"properties": {
"json": true,
"type": {
"type": "string",
"enum": [
"json"
]
}
}
},
{
"type": "object",
"required": [
"arguments",
"call_id",
"name",
"type"
],
"properties": {
"arguments": {
"type": "string"
},
"call_id": {
"type": "string"
},
"name": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"tool_call"
]
}
}
},
{
"type": "object",
"required": [
"call_id",
"output",
"type"
],
"properties": {
"call_id": {
"type": "string"
},
"output": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"tool_result"
]
}
}
},
{
"type": "object",
"required": [
"action",
"path",
"type"
],
"properties": {
"action": {
"$ref": "#/definitions/FileAction"
},
"diff": {
"type": [
"string",
"null"
]
},
"path": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"file_ref"
]
}
}
},
{
"type": "object",
"required": [
"text",
"type",
"visibility"
],
"properties": {
"text": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"reasoning"
]
},
"visibility": {
"$ref": "#/definitions/ReasoningVisibility"
}
}
},
{
"type": "object",
"required": [
"path",
"type"
],
"properties": {
"mime": {
"type": [
"string",
"null"
]
},
"path": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"image"
]
}
}
},
{
"type": "object",
"required": [
"label",
"type"
],
"properties": {
"detail": {
"type": [
"string",
"null"
]
},
"label": {
"type": "string"
},
"type": {
"type": "string",
"enum": [
"status"
]
}
}
}
]
},
"ErrorData": {
"type": "object",
"required": [
"message"
],
"properties": {
"code": {
"type": [
"string",
"null"
]
},
"details": true,
"message": {
"type": "string"
}
}
},
"EventSource": {
"type": "string",
"enum": [
"agent",
"daemon"
]
},
"FileAction": {
"type": "string",
"enum": [
"read",
"write",
"patch"
]
},
"ItemDeltaData": {
"type": "object",
"required": [
"delta",
"item_id"
],
"properties": {
"delta": {
"type": "string"
},
"item_id": {
"type": "string"
},
"native_item_id": {
"type": [
"string",
"null"
]
}
}
},
"ItemEventData": {
"type": "object",
"required": [
"item"
],
"properties": {
"item": {
"$ref": "#/definitions/UniversalItem"
}
}
},
"ItemKind": {
"type": "string",
"enum": [
"message",
"tool_call",
"tool_result",
"system",
"status",
"unknown"
]
},
"ItemRole": {
"type": "string",
"enum": [
"user",
"assistant",
"system",
"tool"
]
},
"ItemStatus": {
"type": "string",
"enum": [
"in_progress",
"completed",
"failed"
]
},
"PermissionEventData": {
"type": "object",
"required": [
"action",
"permission_id",
"status"
],
"properties": {
"action": {
"type": "string"
},
"metadata": true,
"permission_id": {
"type": "string"
},
"status": {
"$ref": "#/definitions/PermissionStatus"
}
}
},
"PermissionStatus": {
"type": "string",
"enum": [
"requested",
"approved",
"denied"
]
},
"QuestionEventData": {
"type": "object",
"required": [
"options",
"prompt",
"question_id",
"status"
],
"properties": {
"options": {
"type": "array",
"items": {
"type": "string"
}
},
"prompt": {
"type": "string"
},
"question_id": {
"type": "string"
},
"response": {
"type": [
"string",
"null"
]
},
"status": {
"$ref": "#/definitions/QuestionStatus"
}
}
},
"QuestionStatus": {
"type": "string",
"enum": [
"requested",
"answered",
"rejected"
]
},
"ReasoningVisibility": {
"type": "string",
"enum": [
"public",
"private"
]
},
"SessionEndReason": {
"type": "string",
"enum": [
"completed",
"error",
"terminated"
]
},
"SessionEndedData": {
"type": "object",
"required": [
"reason",
"terminated_by"
],
"properties": {
"reason": {
"$ref": "#/definitions/SessionEndReason"
},
"terminated_by": {
"$ref": "#/definitions/TerminatedBy"
}
}
},
"SessionStartedData": {
"type": "object",
"properties": {
"metadata": true
}
},
"TerminatedBy": {
"type": "string",
"enum": [
"agent",
"daemon"
]
},
"UniversalEventData": {
"anyOf": [
{
"$ref": "#/definitions/SessionStartedData"
},
{
"$ref": "#/definitions/SessionEndedData"
},
{
"$ref": "#/definitions/ItemEventData"
},
{
"$ref": "#/definitions/ItemDeltaData"
},
{
"$ref": "#/definitions/ErrorData"
},
{
"$ref": "#/definitions/PermissionEventData"
},
{
"$ref": "#/definitions/QuestionEventData"
},
{
"$ref": "#/definitions/AgentUnparsedData"
}
]
},
"UniversalEventType": {
"type": "string",
"enum": [
"session.started",
"session.ended",
"item.started",
"item.delta",
"item.completed",
"error",
"permission.requested",
"permission.resolved",
"question.requested",
"question.resolved",
"agent.unparsed"
]
},
"UniversalItem": {
"type": "object",
"required": [
"content",
"item_id",
"kind",
"status"
],
"properties": {
"content": {
"type": "array",
"items": {
"$ref": "#/definitions/ContentPart"
}
},
"item_id": {
"type": "string"
},
"kind": {
"$ref": "#/definitions/ItemKind"
},
"native_item_id": {
"type": [
"string",
"null"
]
},
"parent_id": {
"type": [
"string",
"null"
]
},
"role": {
"anyOf": [
{
"$ref": "#/definitions/ItemRole"
},
{
"type": "null"
}
]
},
"status": {
"$ref": "#/definitions/ItemStatus"
}
}
}
}
}

View file

@ -2,3 +2,7 @@
- [x] Replace server --mock flag with built-in mock agent and update UI approvals layout.
- [x] Add telemetry module with opt-out flag and sandbox provider detection.
- [x] Add turn-stream message endpoint with SSE response and tests.
- [x] Update CLI + TypeScript SDK/OpenAPI for turn streaming.
- [x] Add inspector UI mode for turn stream and wire send flow.
- [x] Refresh docs for new endpoint and UI mode.