Skip to content

[SYMPP-3107] Retry read-only requests after backend recovery #625

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

Merged
Pimpmuckl merged 6 commits into from
Aug 18, 2026
Merged

[SYMPP-3107] Retry read-only requests after backend recovery #625

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4be1534
fix(mcp): replay reads after backend recovery
Pimpmuckl Aug 18, 2026
c29f800
fix(mcp): block replay for batched tool calls
Pimpmuckl Aug 18, 2026
65ac2b2
fix(mcp): keep recovery scoped to supported HTTP requests
Pimpmuckl Aug 18, 2026
77b63ee
fix(mcp): preserve ambiguous fallback tool calls
Pimpmuckl Aug 18, 2026
5e94920
fix(mcp): classify fallback requests case-sensitively
Pimpmuckl Aug 18, 2026
8bcfb75
fix(mcp): remove fallback request parse limit
Pimpmuckl Aug 18, 2026
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
2 changes: 1 addition & 1 deletion plugins/symphony-plus-plus-mcp/scripts/start-sympp-mcp-bridge.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -873,7 +873,7 @@ async function bridge(identity, state, runtimeFile) {
process.stdout.write(`${JSON.stringify(indeterminateToolCall(parsed.id))}\n`);
continue;
}
if (recovered && replayProvablyUnsent(response)) {
if (recovered && (replayProvablyUnsent(response) || !parsed || parsed.method !== "tools/call")) {
Comment thread
Pimpmuckl marked this conversation as resolved.
try {
if (!requestProtocol) await initializeSession();
response = await mcpPost(current.mcpUrl, line, sessionId, protocol, timeoutMs);
Expand Down
2 changes: 1 addition & 1 deletion plugins/symphony-plus-plus-mcp/scripts/sympp-mcp-process-runtime.ps1
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -712,7 +712,7 @@ function Invoke-HttpMcpBridge([string]$McpUrl, [int]$TimeoutSec, [string]$Client
if ($null -ne $recovered) {
$McpUrl = $recovered.mcp_url; $heartbeatIntervalMs = $recovered.heartbeat_interval_ms
$sessionId = $null; $protocolVersion = $null; $needsInitialize = $true
if (-not $requestMayHaveReachedBackend) {
if (-not $requestMayHaveReachedBackend -or -not (Test-McpToolCall $line)) {
if (-not [string]::IsNullOrWhiteSpace($requestProtocolVersion)) {
$needsInitialize = $false
} else {
Expand Down
35 changes: 21 additions & 14 deletions plugins/symphony-plus-plus-mcp/tests/launcher/cold-start-singleton-smoke.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ function backendFixture() {
return [
'"use strict";',
'const crypto=require("crypto"),fs=require("fs"),http=require("http"),path=require("path");',
'const a=process.argv.slice(2),arg=(n)=>a[a.indexOf(n)+1],port=Number(arg("--port")),stateFile=arg("--state"),release=arg("--release"),bindRelease=arg("--bind-release"),failAfterProbe=arg("--fail-after-probe"),contract=arg("--contract"),revision=arg("--revision"),ledger=arg("--ledger");',
'const a=process.argv.slice(2),arg=(n)=>a[a.indexOf(n)+1],port=Number(arg("--port")),stateFile=arg("--state"),release=arg("--release"),bindRelease=arg("--bind-release"),failAfterProbe=arg("--fail-after-probe"),failRead=arg("--fail-read"),contract=arg("--contract"),revision=arg("--revision"),ledger=arg("--ledger");',
'const tools=JSON.parse(Buffer.from(arg("--tools"),"base64").toString("utf8")),leases=new Set(),sessions=new Set();let failAfterProbeArmed=false;',
'let previous={};try{previous=JSON.parse(fs.readFileSync(stateFile,"utf8"));}catch(_){}',
'const state={pid:process.pid,starts:(previous.starts||0)+1,started_at:Date.now(),initialize:previous.initialize||0,tools_list:previous.tools_list||0,mutations:previous.mutations||0,attach:previous.attach||0,detach:previous.detach||0,lease_peak:previous.lease_peak||0,active_leases:0};',
Expand All @@ -99,15 +99,15 @@ function backendFixture() {
' if(req.url==="/mcp/readiness"){if(release&&!fs.existsSync(release))return send(res,503,{status:"starting"});return send(res,200,{status:"ok",ledger:{reachable:true},dashboard:{ready:true},source:{revision,mcp_contract:{fingerprint:contract}}});}',
' if(req.url==="/sympp/board")return send(res,200,"<title>Symphony++ Dashboard</title>",{"Content-Type":"text/html"});',
' if(req.url==="/mcp/client-lease"){const p=JSON.parse(await body(req));if(p.action==="attach")state.attach++;if(p.action==="attach"||p.action==="heartbeat")leases.add(p.client_id);if(p.action==="heartbeat"&&failAfterProbe&&fs.existsSync(failAfterProbe)){fs.unlinkSync(failAfterProbe);failAfterProbeArmed=true;}if(p.action==="detach"){state.detach++;leases.delete(p.client_id);}state.lease_peak=Math.max(state.lease_peak,leases.size);save();return send(res,200,{stale_after_ms:600000});}',
' if(req.url==="/mcp"){const p=JSON.parse(await body(req));let result,session=req.headers["mcp-session-id"];if(p.method==="initialize"){state.initialize++;session=crypto.randomUUID();sessions.add(session);result={protocolVersion:"2025-03-26",capabilities:{},serverInfo:{name:"cold-fixture",version:"1"}};}else if(!sessions.has(session))return send(res,404,{error:"missing session"});else if(p.method==="tools/list"){state.tools_list++;result={tools:tools.map(name=>({name,description:"fixture",inputSchema:{type:"object"}}))};}else if(p.method==="tools/call"&&p.params.name==="fixture.mutate"){state.mutations++;save();res.writeHead(200,{"Content-Type":"application/json"});res.write("{\\\"jsonrpc\\\":\\\"2.0\\\",\\\"id\\\":");return setTimeout(()=>{res.destroy();setTimeout(()=>process.exit(0),10);},25);}else return send(res,404,{error:"missing"});save();return send(res,200,{jsonrpc:"2.0",id:p.id,result},{"Mcp-Session-Id":session});}',
' if(req.url==="/mcp"){const p=JSON.parse(await body(req));let result,session=req.headers["mcp-session-id"];if(p.method==="initialize"){state.initialize++;session=crypto.randomUUID();sessions.add(session);result={protocolVersion:"2025-03-26",capabilities:{},serverInfo:{name:"cold-fixture",version:"1"}};}else if(!sessions.has(session))return send(res,404,{error:"missing session"});else if(p.method==="tools/list"){state.tools_list++;if(failRead&&fs.existsSync(failRead)){fs.unlinkSync(failRead);save();res.writeHead(200,{"Content-Type":"application/json"});res.write("{\\\"jsonrpc\\\":\\\"2.0\\\",\\\"id\\\":");return setTimeout(()=>{server.close();server.closeAllConnections?.();setTimeout(()=>process.exit(0),10);},25);}result={tools:tools.map(name=>({name,description:"fixture",inputSchema:{type:"object"}}))};}else if(p.method==="tools/call"&&p.params.name==="fixture.mutate"){state.mutations++;save();res.writeHead(200,{"Content-Type":"application/json"});res.write("{\\\"jsonrpc\\\":\\\"2.0\\\",\\\"id\\\":");return setTimeout(()=>{res.destroy();setTimeout(()=>process.exit(0),10);},25);}else return send(res,404,{error:"missing"});save();return send(res,200,{jsonrpc:"2.0",id:p.id,result},{"Mcp-Session-Id":session});}',
' send(res,404,{error:"not found"});',
'});',
'server.on("connection",socket=>{if(!failAfterProbeArmed)return;failAfterProbeArmed=false;socket.on("close",()=>{server.close();server.closeAllConnections?.();setTimeout(()=>process.exit(0),10);});});',
'fs.mkdirSync(path.dirname(ledger),{recursive:true});fs.writeFileSync(ledger,"fixture");function listen(){if(bindRelease&&!fs.existsSync(bindRelease))return setTimeout(listen,25);server.listen(port,"127.0.0.1",save);}listen();'
].join("\n");
}

function createArchive(root, shell, backendPort, backendState, releaseFile, bindReleaseFile, failAfterProbeFile, ledgerFile) {
function createArchive(root, shell, backendPort, backendState, releaseFile, bindReleaseFile, failAfterProbeFile, failReadFile, ledgerFile) {
const source = path.join(root, "artifact-source");
const archive = path.join(root, "artifact.zip");
fs.mkdirSync(path.join(source, "dashboard"), { recursive: true });
Expand All @@ -122,7 +122,7 @@ function createArchive(root, shell, backendPort, backendState, releaseFile, bind
archive,
sha: sha256(fs.readFileSync(archive)),
dashboardHash,
runtimeArgs: ["--port", "{port}", "--state", backendState, "--release", releaseFile, "--bind-release", bindReleaseFile, "--fail-after-probe", failAfterProbeFile, "--contract", contract, "--revision", revision, "--ledger", ledgerFile, "--tools", Buffer.from(JSON.stringify(expectedTools)).toString("base64"), "start-runtime.ps1"],
runtimeArgs: ["--port", "{port}", "--state", backendState, "--release", releaseFile, "--bind-release", bindReleaseFile, "--fail-after-probe", failAfterProbeFile, "--fail-read", failReadFile, "--contract", contract, "--revision", revision, "--ledger", ledgerFile, "--tools", Buffer.from(JSON.stringify(expectedTools)).toString("base64"), "start-runtime.ps1"],
backendPort,
};
}
Expand Down Expand Up @@ -241,6 +241,7 @@ async function runCase(clientCount, shell, mode = "normal") {
const releaseFile = path.join(root, "backend-ready");
const bindReleaseFile = path.join(root, "backend-bind-ready");
const failAfterProbeFile = path.join(root, "fail-after-probe");
const failReadFile = path.join(root, "fail-read");
const ledgerFile = path.join(root, "ledger", "fixture.sqlite3");
const barrier = path.join(root, "barrier");
const backendPort = await freePort();
Expand All @@ -261,7 +262,7 @@ async function runCase(clientCount, shell, mode = "normal") {
writeJson(path.join(sourceRoot, "elixir", "priv", "symphony_plus_plus", "mcp_contract.json"), { mcp_contract_fingerprint: contract });
if (mode !== "backend_death") fs.writeFileSync(releaseFile, "ready");
if (mode !== "backend_prebind_death") fs.writeFileSync(bindReleaseFile, "ready");
const artifact = createArchive(root, shell, backendPort, backendState, releaseFile, bindReleaseFile, failAfterProbeFile, ledgerFile);
const artifact = createArchive(root, shell, backendPort, backendState, releaseFile, bindReleaseFile, failAfterProbeFile, failReadFile, ledgerFile);
let resolvedManifest;
channel = await createChannelServer(mode, () => JSON.stringify(resolvedManifest), artifact.archive);
resolvedManifest = {
Expand All @@ -279,7 +280,7 @@ async function runCase(clientCount, shell, mode = "normal") {
SYMPP_COLD_START_TIMEOUT_SEC: "90", SYMPP_BACKEND_STARTUP_TIMEOUT_SEC: "60", SYMPP_BACKEND_PORT_RELEASE_TIMEOUT_SEC: "1",
SYMPP_ELIXIR_SETUP_TIMEOUT_SEC: "30", SYMPP_AUTOSTART_FRONTEND: "0", SYMPP_MCP_HTTP_TIMEOUT_SEC: "30", TEMP: path.join(root, "tmp"), TMP: path.join(root, "tmp") };
if (mode === "shutdown_during_recovery") environment.SYMPP_MCP_CLIENT_HEARTBEAT_SEC = "5";
if (["powershell_fallback", "powershell_fallback_recovery", "powershell_fallback_initialize_retry"].includes(mode)) environment.SYMPP_NODE_BRIDGE = "0";
if (mode.startsWith("powershell_fallback")) environment.SYMPP_NODE_BRIDGE = "0";
for (const name of ["SYMPP_REPO_ROOT", "SYMPP_BACKEND_URL", "SYMPP_DASHBOARD_ORIGIN", "SYMPP_DATABASE", "SYMPP_SOURCE_FALLBACK", "SYMPP_ARTIFACT_RUNTIME"]) delete environment[name];
fs.mkdirSync(environment.TEMP, { recursive: true });

Expand Down Expand Up @@ -323,6 +324,7 @@ async function runCase(clientCount, shell, mode = "normal") {
}
const firstBackend = readJson(backendState);
const firstOwnerPid = Number(readJson(runtimeFile)?.publication?.owner_adapter_pid || 0);
const backendOnlyReadRecovery = mode.endsWith("backend_only_read_recovery");
let recoveryClients = clients;
if (mode === "powershell_fallback_initialize_retry") {
fs.writeFileSync(failAfterProbeFile, "ready");
Expand Down Expand Up @@ -441,6 +443,10 @@ async function runCase(clientCount, shell, mode = "normal") {
await terminate(firstBackend.pid);
await waitFor(() => clients.filter((client) => client.child.exitCode === null).length === clientCount - 1, "Backend owner adapter did not exit.");
recoveryClients = clients.filter((client) => client.child.exitCode === null);
} else if (backendOnlyReadRecovery) {
fs.writeFileSync(failReadFile, "ready");
const recovered = await requestClient(clients[0], 3500, "tools/list");
assert.equal(recovered.result?.tools?.length, expectedTools.length, `Read-only request was not retried after backend exit. ${JSON.stringify(recovered)}`);
}
if (["backend_loss", "owner_loss"].includes(mode)) {
const recovered = await Promise.all(recoveryClients.map((client, index) => requestClient(client, 1000 + index, "tools/list")));
Expand All @@ -452,8 +458,8 @@ async function runCase(clientCount, shell, mode = "normal") {
const recovered = await requestClient(clients[0], 2001, "tools/list");
assert.equal(recovered.result?.tools?.length, expectedTools.length);
}
if (["backend_loss", "owner_loss", "ambiguous_tool"].includes(mode)) {
const expectedRecoveredLeases = mode === "ambiguous_tool" ? 1 : recoveryClients.length;
if (["backend_loss", "owner_loss", "ambiguous_tool"].includes(mode) || backendOnlyReadRecovery) {
const expectedRecoveredLeases = mode === "ambiguous_tool" || backendOnlyReadRecovery ? 1 : recoveryClients.length;
await waitFor(() => readJson(backendState)?.active_leases === expectedRecoveredLeases, "Recovered adapters did not retain their replacement backend leases.");
}
const activeBackend = readJson(backendState);
Expand All @@ -471,11 +477,11 @@ async function runCase(clientCount, shell, mode = "normal") {
await waitFor(() => portAvailable(backendPort), "Backend listener did not stop after the final client exited.");
const state = await waitFor(() => { const value = readJson(runtimeFile); return value?.backend?.status === "stopped" && value.backend.pid === null && value; }, "Runtime state did not record zero-client backend shutdown.");
assert.equal(state.publication.status, "ready");
const recoveryMode = ["backend_loss", "owner_loss", "ambiguous_tool"].includes(mode);
const recoveryRebinds = ["backend_loss", "owner_loss"].includes(mode) ? recoveryClients.length : mode === "ambiguous_tool" ? 1 : 0;
const recoveryMode = ["backend_loss", "owner_loss", "ambiguous_tool"].includes(mode) || backendOnlyReadRecovery;
const recoveryRebinds = ["backend_loss", "owner_loss"].includes(mode) ? recoveryClients.length : mode === "ambiguous_tool" || backendOnlyReadRecovery ? 1 : 0;
assert.equal(backend.starts, recoveryMode ? 2 : 1);
assert.equal(backend.initialize, clientCount + recoveryRebinds);
assert.equal(backend.tools_list, clientCount + recoveryRebinds);
assert.equal(backend.tools_list, clientCount + recoveryRebinds + (backendOnlyReadRecovery ? 1 : 0));
assert.equal(backend.mutations, mode === "ambiguous_tool" ? 1 : 0);
assert.equal(backend.lease_peak, clientCount);
assert.equal(backend.active_leases, 0);
Expand All @@ -484,7 +490,8 @@ async function runCase(clientCount, shell, mode = "normal") {
assert.equal(channel.counts.archive_successes, 1);
assert.equal(traceCount(traceDir, "artifact_prepare_end"), 1);
assert.equal(traceCount(traceDir, "runtime_ready_published"), recoveryMode ? 2 : 1);
assert.equal(traceCount(traceDir, "backend_recovery_leader"), recoveryMode ? 1 : 0);
const recoveryLeaders = backendOnlyReadRecovery && mode.startsWith("powershell_fallback") ? traceCount(traceDir, "runtime_ready_published") - 1 : traceCount(traceDir, "backend_recovery_leader");
assert.equal(recoveryLeaders, recoveryMode ? 1 : 0);
if (recoveryMode) {
assert.notEqual(activeBackend.pid, firstBackend.pid);
}
Expand All @@ -503,7 +510,7 @@ async function runCase(clientCount, shell, mode = "normal") {
for (const directory of [symppHome, environment.TEMP]) if (fs.existsSync(directory)) for (const entry of fs.readdirSync(directory, { recursive: true })) if (/artifact\.zip\.tmp-|\.extracting-|codex-plugin\.json\.tmp-/.test(String(entry))) leftovers.push(entry);
assert.deepEqual(leftovers, []);
assert.ok(percentile(latencies, 0.95) < 60000 && Math.max(...latencies) < 90000);
return { mode, shell: path.basename(shell), clients: clientCount, p95_ms: percentile(latencies, 0.95), max_ms: Math.max(...latencies), manifest: channel.counts.manifest_successes, manifest_attempts: channel.counts.manifest_attempts, artifact: channel.counts.archive_successes, artifact_attempts: channel.counts.archive_attempts, preparations: traceCount(traceDir, "artifact_prepare_end"), backends: backend.starts, pids: recoveryMode ? 2 : 1, listeners: 0, initializes: backend.initialize, tools_list: backend.tools_list, mutations: backend.mutations, lease_peak: backend.lease_peak, leases_after: backend.active_leases, adopted: traceCount(traceDir, "backend_adopted"), recovery_leaders: traceCount(traceDir, "backend_recovery_leader") };
return { mode, shell: path.basename(shell), clients: clientCount, p95_ms: percentile(latencies, 0.95), max_ms: Math.max(...latencies), manifest: channel.counts.manifest_successes, manifest_attempts: channel.counts.manifest_attempts, artifact: channel.counts.archive_successes, artifact_attempts: channel.counts.archive_attempts, preparations: traceCount(traceDir, "artifact_prepare_end"), backends: backend.starts, pids: recoveryMode ? 2 : 1, listeners: 0, initializes: backend.initialize, tools_list: backend.tools_list, mutations: backend.mutations, lease_peak: backend.lease_peak, leases_after: backend.active_leases, adopted: traceCount(traceDir, "backend_adopted"), recovery_leaders: recoveryLeaders };
} finally {
terminateTrees(clients.filter((client) => client.child.exitCode === null).map((client) => client.child.pid));
if (!backendPid) backendPid = readJson(backendState)?.pid || 0;
Expand All @@ -529,7 +536,7 @@ async function main() {
const powershellFallback = await runCase(30, windowsPowerShell, "powershell_fallback");
for (const mode of ["manifest_death", "artifact_death", "backend_death", "backend_prebind_death"]) results.push(await runCase(30, pwsh, mode));
const recovery = [];
for (const mode of ["owner_loss", "backend_loss", "ambiguous_tool", "shutdown_during_recovery", "generation_changed_recovery", "cleanup_source_changed_recovery", "powershell_fallback_recovery", "powershell_fallback_initialize_retry"]) recovery.push(await runCase(["shutdown_during_recovery", "generation_changed_recovery", "cleanup_source_changed_recovery"].includes(mode) ? 3 : mode === "powershell_fallback_recovery" ? 4 : mode === "powershell_fallback_initialize_retry" ? 1 : 10, mode.startsWith("powershell_fallback") ? windowsPowerShell : pwsh, mode));
for (const mode of ["owner_loss", "backend_loss", "backend_only_read_recovery", "ambiguous_tool", "shutdown_during_recovery", "generation_changed_recovery", "cleanup_source_changed_recovery", "powershell_fallback_recovery", "powershell_fallback_backend_only_read_recovery", "powershell_fallback_initialize_retry"]) recovery.push(await runCase(["shutdown_during_recovery", "generation_changed_recovery", "cleanup_source_changed_recovery"].includes(mode) ? 3 : mode === "powershell_fallback_recovery" ? 4 : mode.endsWith("backend_only_read_recovery") || mode === "powershell_fallback_initialize_retry" ? 1 : 10, mode.startsWith("powershell_fallback") ? windowsPowerShell : pwsh, mode));
process.stdout.write(`${JSON.stringify({ matrix: results.slice(0, 3), powershell_fallback: powershellFallback, leader_death: results.slice(3), recovery, powershell_5_1: true, pwsh: true, cleanup: true })}\n`);
}

Expand Down
Loading