Skip to content

fix(agents): persist and resume native sessions #1742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
NathanFlurry wants to merge 1 commit into
base: main
Choose a base branch
from
+832 −122
Draft

fix(agents): persist and resume native sessions #1742

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 44 additions & 12 deletions crates/agentos-sidecar-core/src/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1098,24 +1098,36 @@ fn native_resume_method(agent_capabilities: Option<&Value>) -> Option<&'static s
None
}

/// Normalize an adapter "no such session" error (`-32603` + `details ==
/// "NotFoundError"`) into the shared `unknown_session` discriminator. Strict on
/// purpose: malformed `session/load` must still propagate. Mirrors the native
/// `normalize_unknown_session_error`.
/// Normalize the exact missing-session shapes emitted by OpenCode and pi-acp
/// into the shared `unknown_session` discriminator. Strict on purpose: malformed
/// `session/load` must still propagate. Mirrors the native sidecar implementation.
fn normalize_unknown_session_error(response: &mut Value) {
let Some(error) = response.get_mut("error").and_then(Value::as_object_mut) else {
return;
};
let code = error.get("code").and_then(Value::as_i64);
let Some(data) = error.get_mut("data").and_then(Value::as_object_mut) else {
if code == Some(-32603) {
let Some(data) = error.get_mut("data").and_then(Value::as_object_mut) else {
return;
};
if data.get("details").and_then(Value::as_str) == Some("NotFoundError") {
data.insert(
String::from("kind"),
Value::String(String::from("unknown_session")),
);
}
return;
};
let details = data.get("details").and_then(Value::as_str);
if code == Some(-32603) && details == Some("NotFoundError") {
data.insert(
String::from("kind"),
Value::String(String::from("unknown_session")),
);
}
if code == Some(-32602) {
let Some(details) = error.get("data").and_then(Value::as_str) else {
return;
};
if details.starts_with("Unknown sessionId:") {
error.insert(
String::from("data"),
serde_json::json!({ "kind": "unknown_session", "details": details }),
);
}
}
}

Expand Down Expand Up @@ -1256,6 +1268,26 @@ mod tests {
use super::*;
use crate::host::{AgentOutput, SpawnAgentRequest, SpawnedAgent};

#[test]
fn unknown_session_normalization_pins_adapter_shapes() {
let mut opencode = json!({
"error": { "code": -32603, "data": { "details": "NotFoundError" } }
});
normalize_unknown_session_error(&mut opencode);
assert!(is_unknown_session_error(&opencode));

let mut pi_acp = json!({
"error": { "code": -32602, "data": "Unknown sessionId: missing" }
});
normalize_unknown_session_error(&mut pi_acp);
assert!(is_unknown_session_error(&pi_acp));

let mut malformed = json!({
"error": { "code": -32602, "data": { "sessionId": ["expected string"] } }
});
normalize_unknown_session_error(&mut malformed);
assert!(!is_unknown_session_error(&malformed));
}
#[derive(Default)]
struct MockHost {
killed: Vec<(String, String)>,
Expand Down
60 changes: 44 additions & 16 deletions crates/agentos-sidecar/src/acp_extension.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2788,27 +2788,37 @@ fn trace_acp_response(method: &str, response: &Value) {
/// Normalize adapter-specific "no such session" errors from `session/load` into
/// the shared `unknown_session` discriminator used by the resume state machine.
///
/// OpenCode currently reports a missing session as JSON-RPC `-32603` with
/// `error.data.details == "NotFoundError"`: its ACP server converts thrown
/// non-`RequestError` exceptions into `internalError({ details: error.message })`,
/// and `Session.get` throws a `NotFoundError` whose message is the class name.
/// Convert exactly that shape into `error.data.kind = "unknown_session"` before
/// fallback matching. Do not broaden this to message substrings or all
/// `-32603`/`-32602` errors; malformed `session/load` must still propagate.
/// OpenCode reports `-32603` plus `{ details: "NotFoundError" }`. pi-acp 0.0.23
/// reports `-32602` with a string data value beginning `Unknown sessionId:`.
/// Normalize only those exact adapter shapes; malformed load requests and other
/// internal errors must still propagate.
fn normalize_unknown_session_error(response: &mut Value) {
let Some(error) = response.get_mut("error").and_then(Value::as_object_mut) else {
return;
};
let code = error.get("code").and_then(Value::as_i64);
let Some(data) = error.get_mut("data").and_then(Value::as_object_mut) else {
if code == Some(-32603) {
let Some(data) = error.get_mut("data").and_then(Value::as_object_mut) else {
return;
};
if data.get("details").and_then(Value::as_str) == Some("NotFoundError") {
data.insert(
String::from("kind"),
Value::String(String::from("unknown_session")),
);
}
return;
};
let details = data.get("details").and_then(Value::as_str);
if code == Some(-32603) && details == Some("NotFoundError") {
data.insert(
String::from("kind"),
Value::String(String::from("unknown_session")),
);
}
if code == Some(-32602) {
let Some(details) = error.get("data").and_then(Value::as_str) else {
return;
};
if details.starts_with("Unknown sessionId:") {
error.insert(
String::from("data"),
serde_json::json!({ "kind": "unknown_session", "details": details }),
);
}
}
}

Expand Down Expand Up @@ -3100,7 +3110,7 @@ mod tests {
}

#[test]
fn unknown_session_normalization_pins_opencode_shape() {
fn unknown_session_normalization_pins_adapter_shapes() {
let mut opencode = serde_json::json!({
"error": { "code": -32603, "message": "Internal error", "data": { "details": "NotFoundError" } }
});
Expand All @@ -3111,13 +3121,31 @@ mod tests {
);
assert!(is_unknown_session_error(&opencode));

let mut pi_acp = serde_json::json!({
"error": { "code": -32602, "message": "Invalid params",
"data": "Unknown sessionId: missing-pi-session" }
});
normalize_unknown_session_error(&mut pi_acp);
assert_eq!(
pi_acp.pointer("/error/data/kind").and_then(Value::as_str),
Some("unknown_session")
);
assert!(is_unknown_session_error(&pi_acp));

let mut malformed = serde_json::json!({
"error": { "code": -32602, "message": "Invalid params",
"data": { "_errors": [], "sessionId": { "_errors": ["expected string"] } } }
});
normalize_unknown_session_error(&mut malformed);
assert!(!is_unknown_session_error(&malformed));

let mut other_invalid_params = serde_json::json!({
"error": { "code": -32602, "message": "Invalid params",
"data": "cwd must be an absolute path" }
});
normalize_unknown_session_error(&mut other_invalid_params);
assert!(!is_unknown_session_error(&other_invalid_params));

let mut other_internal = serde_json::json!({
"error": { "code": -32603, "message": "Internal error", "data": { "details": "SomethingElse" } }
});
Expand Down
69 changes: 48 additions & 21 deletions packages/core/tests/agent-pkg-matrix.e2e.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
* For every package manager (npm/pnpm/yarn/bun) × every agent
* (pi/pi-cli/claude/opencode) this installs the PUBLISHED packages into an
* isolated temp project and asserts a real user can: install → create a session
* → prompt → stream tokens LIVE. It is the regression net for the exact issues
* → prompt → stream tokens LIVE → native resume → transcript restore. It is the
* regression net for the exact issues
* that bit us shipping the preview:
*
* - retired/stale model ids (Anthropic 404 → empty turn),
Expand All @@ -30,6 +31,9 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
* ANTHROPIC_API_KEY required
* AGENTOS_MATRIX_CORE @rivet-dev/agentos-core version/tag (default "latest")
* AGENTOS_MATRIX_AGENTS @agentos-software/* version/tag (default "latest")
* AGENTOS_MATRIX_CORE_SPEC exact install spec, e.g. file:/path/to/packages/core
* AGENTOS_MATRIX_<AGENT>_SPEC exact per-agent install spec; PI_CLI uses an underscore
* AGENTOS_MATRIX_REPO_ROOT import built artifacts from this checkout; skips installation
* AGENTOS_MATRIX_MODEL opencode model id (default a current Haiku)
* AGENTOS_MATRIX_PMS comma list to restrict package managers
* AGENTOS_MATRIX_AGENTS_LIST comma list to restrict agents
Expand All @@ -40,6 +44,7 @@ const CORE_VERSION = process.env.AGENTOS_MATRIX_CORE || "latest";
const AGENTS_VERSION = process.env.AGENTOS_MATRIX_AGENTS || "latest";
const CELL = resolve(import.meta.dirname, "fixtures/agent-matrix-cell.mjs");
const CELL_TIMEOUT_MS = 240_000;
const REPO_ROOT = process.env.AGENTOS_MATRIX_REPO_ROOT;

const AGENT_PKGS: Record<string, string> = {
pi: "@agentos-software/pi",
Expand All @@ -48,6 +53,11 @@ const AGENT_PKGS: Record<string, string> = {
opencode: "@agentos-software/opencode",
};

function agentInstallSpec(agent: string): string {
const envName = `AGENTOS_MATRIX_${agent.toUpperCase().replaceAll("-", "_")}_SPEC`;
return process.env[envName] || `${AGENT_PKGS[agent]}@${AGENTS_VERSION}`;
}

function commandAvailable(cmd: string): boolean {
try {
const r = spawnSync(cmd, ["--version"], { stdio: "ignore" });
Expand Down Expand Up @@ -97,6 +107,7 @@ const ALL_AGENTS = (

const availablePms = ALL_PMS.filter(commandAvailable);

// biome-ignore format: keep the matrix definition readable with inline Vitest options.
describe.skipIf(!ENABLED)("agent × package-manager e2e matrix (real API)", () => {
const tmpDirs: string[] = [];

Expand Down Expand Up @@ -130,13 +141,14 @@ describe.skipIf(!ENABLED)("agent × package-manager e2e matrix (real API)", () =
for (const pm of availablePms) {
for (const agent of ALL_AGENTS) {
it(
`${pm} + ${agent}: install → session → live token streaming`,
// opencode's ACP bootstrap (and real LLM APIs) flake transiently;
// retry the whole cell so a flake doesn't red the gate. Persistent
// failures still fail after the retries.
`${pm} + ${agent}: install → stream → native resume → transcript restore`,
// OpenCode's ACP bootstrap and real LLM APIs can flake transiently.
// Retry the whole cell; persistent failures still fail the gate.
{ timeout: CELL_TIMEOUT_MS + 200_000, retry: 2 },
async () => {
const dir = mkdtempSync(join(tmpdir(), `agentos-matrix-${pm}-${agent}-`));
const dir = mkdtempSync(
join(tmpdir(), `agentos-matrix-${pm}-${agent}-`),
);
tmpDirs.push(dir);
// yarn 1.x global cache contends under repeated runs; isolate it.
const cacheDir = join(dir, ".pm-cache");
Expand All @@ -147,17 +159,20 @@ describe.skipIf(!ENABLED)("agent × package-manager e2e matrix (real API)", () =
npm_config_cache: cacheDir,
};

const pkgs = [
`@rivet-dev/agentos-core@${CORE_VERSION}`,
`${AGENT_PKGS[agent]}@${AGENTS_VERSION}`,
];
for (const [cmd, args] of installArgs(pm, pkgs)) {
execFileSync(cmd, args, {
cwd: dir,
env: childEnv,
stdio: "pipe",
timeout: 180_000,
});
if (!REPO_ROOT) {
const pkgs = [
process.env.AGENTOS_MATRIX_CORE_SPEC ||
`@rivet-dev/agentos-core@${CORE_VERSION}`,
agentInstallSpec(agent),
];
for (const [cmd, args] of installArgs(pm, pkgs)) {
execFileSync(cmd, args, {
cwd: dir,
env: childEnv,
stdio: "pipe",
timeout: 180_000,
});
}
}

cpSync(CELL, join(dir, "agent-matrix-cell.mjs"));
Expand All @@ -180,15 +195,27 @@ describe.skipIf(!ENABLED)("agent × package-manager e2e matrix (real API)", () =
const result = JSON.parse(line.slice("E2E_RESULT_JSON:".length));

// eslint-disable-next-line no-console
console.log(`[matrix] ${pm}/${agent}:`, JSON.stringify(result.metrics));

expect(result.ok, `prompt produced output (err: ${result.error})`).toBe(
true,
console.log(
`[matrix] ${pm}/${agent}:`,
JSON.stringify({ metrics: result.metrics, error: result.error }),
);

expect(
result.ok,
`prompt produced output (err: ${result.error})`,
).toBe(true);
expect(
result.streaming,
`tokens streamed live (metrics: ${JSON.stringify(result.metrics)})`,
).toBe(true);
expect(
result.nativeResume,
`native session state resumed (metrics: ${JSON.stringify(result.metrics)})`,
).toBe(true);
expect(
result.transcriptRestore,
`missing native state restored from transcript (metrics: ${JSON.stringify(result.metrics)})`,
).toBe(true);
},
);
}
Expand Down
Loading
Loading