mirror of
https://github.com/dtzp555-max/ocp.git
synced 2026-07-19 09:44:07 +00:00
* docs(spec): design for #47 SSE heartbeat on streaming path Draft spec for an opt-in idle-watchdog SSE heartbeat covering both pre-first-byte and mid-stream silent windows. Default disabled, controlled by CLAUDE_HEARTBEAT_INTERVAL. Targets ~40 LOC. Decisions captured: D1 whole-stream reset-on-byte; D2 SSE comment frame; D3 default disabled; D4 relocate ensureHeaders() to post-spawn; D5 X-Accel-Buffering: no on both SSE header sites; D6 single log line per affected request. Scope-locked: does not touch CLAUDE_TIMEOUT semantics, the separate server.mjs:480-489 dangling-client bug, issues #41/#42, or the non-streaming path. Includes privacy preflight and cloud-testing plan per maintainer feedback. Refs: #47 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * docs(plan): implementation plan for #47 SSE heartbeat 6 phases: pre-work (file 480-bug), implementation (5 tasks on feat/47-sse-heartbeat), opus fresh-context review, cloud verification on Mac rig, privacy preflight, PR+release. Each implementation task carries concrete code, syntax check, and a scoped commit message. LOC budget enforced in Task 1.6 gate (~45 server.mjs lines max). Reviewer checklist scopes scope-lock, ALIGNMENT, privacy, and heartbeat-cannot-abort discipline. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(server): add startHeartbeat helper + HEARTBEAT_INTERVAL env var Per design doc (refs #47). Helper is a per-request idle watchdog that emits `: keepalive\n\n` SSE comment frames; returns a {reset, stop} handle. No wiring yet — helper is unused, safe to commit in isolation. cli.js citation: N/A — SSE response shaping is an OCP-owned translation layer, not a cli.js operation. See AGENTS.md and ALIGNMENT.md Rule 2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(server): eagerly send SSE headers post-spawn (D4, refs #47) Moves the ensureHeaders() call from "on first stdout byte" to "immediately after successful spawn." This is a prerequisite for the heartbeat covering the pre-first-byte silent window (the 'processing large contexts' failure mode in #47). Behavioral consequence: the narrow "spawn succeeded but subprocess died before any byte" branch at server.mjs:610-611 becomes effectively dead in the common case. The post-headers SSE-stop path (612-619) handles it instead. The branch remains defensively for the client-closed-before- ensureHeaders race. Isolated commit per design doc §D4 so reviewer can focus on this one behavior change. cli.js citation: N/A — SSE header emission is OCP response-shaping. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(server): wire heartbeat into streaming path (refs #47) - sendSSE() accepts optional hb handle and calls hb.reset() before write - callClaudeStreaming starts heartbeat after ensureHeaders() and passes hb to the three streaming sendSSE call sites - All three exit paths (proc close, proc error, res close) call hb.stop() to guarantee timer cleanup; no-op handle when disabled means zero runtime cost when CLAUDE_HEARTBEAT_INTERVAL=0 Heartbeat never aborts — only writes comment frames and re-arms. Aligns with v3.3 timeout discipline (single CLAUDE_TIMEOUT, no secondary client-killing timers). cli.js citation: N/A — SSE response shaping is OCP translation layer. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(server): add X-Accel-Buffering: no to SSE response headers (refs #47) nginx (and many LBs / Cloudflare) default to proxy_buffering=on, which would buffer heartbeat comment frames indefinitely and defeat the feature silently. This header hints no-buffering; other stacks ignore it. Applied at both SSE header sites (real streaming + cache-hit). cli.js citation: N/A — response header shaping is OCP translation layer. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * chore(release): v3.12.0 — streaming heartbeat (refs #47) Bundles the release-kit companion files per Iron Rule 5.2 / 11 example: version bump across package.json + ocp-plugin + openclaw.plugin.json, CHANGELOG v3.12.0 section, README env var row + "Streaming heartbeat" explainer. Tag push to v3.12.0 triggers .github/workflows/release.yml to create the GitHub Release automatically. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(server): ensureHeaders returns true when already sent (refs #47) Phase 3 smoke test revealed every content chunk was being dropped after the D4 eager ensureHeaders() call: the stdout.on('data') guard `if (!ensureHeaders()) return;` early-returned on every chunk because ensureHeaders returned false for the already-sent case (conflated with the dead-connection case). Split the two conditions explicitly: return false only when res is ended/destroyed; return true when headers are (already or now) sent. This also fixes a latent multi-chunk bug on main that was masked because claude CLI typically outputs in one stdout chunk. Verified: node -c server.mjs; subsequent re-run of Phase 3 smoke test now sees streaming content chunks + heartbeat frames. cli.js citation: N/A — SSE response shaping is OCP translation layer. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
dtzp555-max
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
cff06439fa
commit
b871b72b6b
+50
-6
@@ -25,6 +25,7 @@
|
||||
* CLAUDE_BREAKER_WINDOW — sliding window duration in ms (default: 300000 = 5min)
|
||||
* CLAUDE_BREAKER_HALF_OPEN_MAX — max concurrent probes in half-open state (default: 2)
|
||||
* PROXY_API_KEY — Bearer token for API auth (optional)
|
||||
* CLAUDE_HEARTBEAT_INTERVAL — SSE heartbeat interval in ms on streaming path (default: 0 = disabled)
|
||||
*/
|
||||
import { createServer } from "node:http";
|
||||
import { spawn, execFileSync } from "node:child_process";
|
||||
@@ -94,6 +95,7 @@ const BREAKER_THRESHOLD = parseInt(process.env.CLAUDE_BREAKER_THRESHOLD || "6",
|
||||
const BREAKER_COOLDOWN = parseInt(process.env.CLAUDE_BREAKER_COOLDOWN || "120000", 10);
|
||||
const BREAKER_WINDOW = parseInt(process.env.CLAUDE_BREAKER_WINDOW || "300000", 10);
|
||||
const BREAKER_HALF_OPEN_MAX = parseInt(process.env.CLAUDE_BREAKER_HALF_OPEN_MAX || "2", 10);
|
||||
const HEARTBEAT_INTERVAL = parseInt(process.env.CLAUDE_HEARTBEAT_INTERVAL || "0", 10);
|
||||
const BIND_ADDRESS = process.env.CLAUDE_BIND || "127.0.0.1";
|
||||
const NO_CONTEXT = process.env.CLAUDE_NO_CONTEXT === "true";
|
||||
const AUTH_MODE = process.env.CLAUDE_AUTH_MODE || (PROXY_API_KEY ? "shared" : "none");
|
||||
@@ -542,6 +544,33 @@ function callClaude(model, messages, conversationId) {
|
||||
});
|
||||
}
|
||||
|
||||
// ── SSE heartbeat (opt-in idle watchdog) ────────────────────────────────
|
||||
// Emits `: keepalive\n\n` SSE comment frames during silent windows on the
|
||||
// streaming response. Design: docs/superpowers/specs/2026-04-25-47-sse-heartbeat-design.md
|
||||
// This is a downstream liveness hint only — it MUST NOT be able to abort
|
||||
// or time out a request. That discipline is load-bearing: v2.2-v2.5's
|
||||
// first-byte/adaptive-tier timeouts "repeatedly killed valid requests"
|
||||
// (see server.mjs top-of-file comment and commit 3843ec8).
|
||||
function startHeartbeat(res, intervalMs, sessionId) {
|
||||
if (!intervalMs || intervalMs <= 0) return { reset: () => {}, stop: () => {} };
|
||||
let handle = null;
|
||||
let hasFired = false;
|
||||
const onFire = () => {
|
||||
if (res.writableEnded || res.destroyed) return;
|
||||
res.write(": keepalive\n\n");
|
||||
if (!hasFired) {
|
||||
hasFired = true;
|
||||
logEvent("info", "heartbeat_active", { session: sessionId, intervalMs });
|
||||
}
|
||||
handle = setTimeout(onFire, intervalMs);
|
||||
};
|
||||
handle = setTimeout(onFire, intervalMs);
|
||||
return {
|
||||
reset: () => { if (handle) { clearTimeout(handle); handle = setTimeout(onFire, intervalMs); } },
|
||||
stop: () => { if (handle) { clearTimeout(handle); handle = null; } },
|
||||
};
|
||||
}
|
||||
|
||||
// ── Call claude CLI (real streaming) ─────────────────────────────────────
|
||||
// Pipes stdout from the claude process directly to SSE chunks as they arrive.
|
||||
// Each data chunk becomes a proper SSE event with delta content in real time.
|
||||
@@ -563,12 +592,14 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
let cachedContent = ""; // accumulate for cache write-back
|
||||
|
||||
function ensureHeaders() {
|
||||
if (headersSent || res.writableEnded || res.destroyed) return false;
|
||||
if (res.writableEnded || res.destroyed) return false;
|
||||
if (headersSent) return true;
|
||||
headersSent = true;
|
||||
res.writeHead(200, {
|
||||
"Content-Type": "text/event-stream",
|
||||
"Cache-Control": "no-cache",
|
||||
"Connection": "keep-alive",
|
||||
"X-Accel-Buffering": "no",
|
||||
});
|
||||
// Send initial role chunk
|
||||
sendSSE(res, {
|
||||
@@ -578,6 +609,15 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
return true;
|
||||
}
|
||||
|
||||
// D4 (spec 2026-04-25): eagerly send SSE headers post-spawn so the
|
||||
// heartbeat started in the next statement (Task 1.3) covers the
|
||||
// pre-first-byte silent window. Behavior change: the `code !== 0`
|
||||
// before-first-byte branch at server.mjs:610-611 becomes effectively
|
||||
// unreachable in the common case — the post-headers SSE-stop path
|
||||
// (612-619) handles it instead.
|
||||
ensureHeaders();
|
||||
const hb = startHeartbeat(res, HEARTBEAT_INTERVAL, convId);
|
||||
|
||||
proc.stdout.on("data", (d) => {
|
||||
markFirstByte();
|
||||
const text = d.toString();
|
||||
@@ -590,13 +630,14 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
sendSSE(res, {
|
||||
id, object: "chat.completion.chunk", created, model,
|
||||
choices: [{ index: 0, delta: { content: text }, finish_reason: null }],
|
||||
});
|
||||
}, hb);
|
||||
});
|
||||
|
||||
proc.stderr.on("data", (d) => (stderr += d));
|
||||
|
||||
proc.on("close", (code, signal) => {
|
||||
activeProcesses.delete(proc);
|
||||
hb.stop();
|
||||
cleanup();
|
||||
const elapsed = Date.now() - t0;
|
||||
|
||||
@@ -613,7 +654,7 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
sendSSE(res, {
|
||||
id, object: "chat.completion.chunk", created, model,
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
});
|
||||
}, hb);
|
||||
res.write("data: [DONE]\n\n");
|
||||
res.end();
|
||||
}
|
||||
@@ -632,7 +673,7 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
sendSSE(res, {
|
||||
id, object: "chat.completion.chunk", created, model,
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
});
|
||||
}, hb);
|
||||
res.write("data: [DONE]\n\n");
|
||||
res.end();
|
||||
}
|
||||
@@ -641,6 +682,7 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
|
||||
proc.on("error", (err) => {
|
||||
console.error(`[claude] spawn error: ${err.message}`);
|
||||
hb.stop();
|
||||
cleanup();
|
||||
trackError(err.message);
|
||||
handleSessionFailure();
|
||||
@@ -653,6 +695,7 @@ function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}
|
||||
|
||||
// If client disconnects, kill the process to free resources
|
||||
res.on("close", () => {
|
||||
hb.stop();
|
||||
if (!proc.killed) {
|
||||
try { proc.kill("SIGTERM"); } catch {}
|
||||
}
|
||||
@@ -666,7 +709,8 @@ function jsonResponse(res, status, data) {
|
||||
res.end(JSON.stringify(data));
|
||||
}
|
||||
|
||||
function sendSSE(res, data) {
|
||||
function sendSSE(res, data, hb) {
|
||||
hb?.reset();
|
||||
res.write(`data: ${JSON.stringify(data)}\n\n`);
|
||||
}
|
||||
|
||||
@@ -1168,7 +1212,7 @@ async function handleChatCompletions(req, res) {
|
||||
// Simulate streaming for cached response
|
||||
const id = `chatcmpl-${randomUUID()}`;
|
||||
const created = Math.floor(Date.now() / 1000);
|
||||
res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive" });
|
||||
res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" });
|
||||
sendSSE(res, { id, object: "chat.completion.chunk", created, model, choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }] });
|
||||
sendSSE(res, { id, object: "chat.completion.chunk", created, model, choices: [{ index: 0, delta: { content: cached.response }, finish_reason: null }] });
|
||||
sendSSE(res, { id, object: "chat.completion.chunk", created, model, choices: [{ index: 0, delta: {}, finish_reason: "stop" }] });
|
||||
|
||||
Reference in New Issue
Block a user