src/server/claude.ts
Rama main · réplica de lectura del repositorio original.
import { mkdirSync, readFileSync, writeFileSync, existsSync } from 'node:fs';
import { summarizeText } from './summarize.ts';
import { dataSync } from './data-sync.ts';
import type { McpConfig } from './mcp.ts';
import type { ClaudeAccountRef } from './claude-account.ts';
import { loadConversation, saveConversation, appendMessage, getSession, setSessionDirectives, addTriggeredSkills, upsertSession, setClaudeResume, getSessionFloor, lowerSessionFloor, type ConvMessage, type ConvBlock } from './sessions.ts';
import { enforcing, TurnGuard } from './security/enforce.ts';
import { createTurnAccumulator, type TurnStreamHooks } from './turn-stream.ts';
import {
resolveDefaultSkillsContent,
expandMentionedSkills,
buildMcpSkillHintsBlock,
buildSkillIndexBlock,
matchTriggeredSkillNames,
skillInjectionBlocks,
getSkill,
getMcpCommandPrompt,
parseSkillFrontmatter,
resolveSkillTurns,
} from './skills.ts';
import { buildWorkspaceContextBlock, expandWorkspaceMentions } from './workspace.ts';
import { buildOffloadContextBlock } from './offload.ts';
import { parseDirectives, type Directives } from './directives.ts';
import { parseSlashCommand, formatCommandBlock } from './commands.ts';
import { getUserContextBlock } from './user-context.ts';
import { collectTurnContext } from './turn-context.ts';
import { DATA_DIR, dataPath } from './paths.ts';
import * as contacts from './contacts.ts';
import type { Principal } from './security/principal.ts';
import { security, resolvePrincipal } from './security/runtime.ts';
import { resolveAndGetEngine, ModelUnavailableError } from './engine/index.ts';
// ── Agent config (shared across users) ──────────────────────────────────────
// The READ path lives in agent-config.ts (a leaf the security runtime can import without pulling this
// module's graph); re-exported here so existing importers keep using claude.ts.
export type { AgentSettings as AgentConfig } from './shraga-config.ts';
import type { AgentSettings as AgentConfig } from './shraga-config.ts';
import { CONFIG_PATH, getAgentConfig } from './agent-config.ts';
import { claudeForcesSubscription } from './claude-account.ts';
export { getAgentConfig };
/**
* Which credentials the claude-code SDK will resolve — process-global, derived from env at call time
* (an API-key env var wins over a stored claude.ai login, matching the SDK's own precedence). Surfaced
* to the UI so the header shows subscription-vs-API-key immediately, without waiting for a turn. The
* per-turn ground truth (SDK `apiKeySource`) is logged by the engine.
*/
export function getClaudeAuthSource(): 'subscription' | 'api-key' {
if (claudeForcesSubscription()) return 'subscription';
return process.env.ANTHROPIC_API_KEY || process.env.ANTHROPIC_AUTH_TOKEN ? 'api-key' : 'subscription';
}
export function saveAgentConfig(config: AgentConfig): void {
mkdirSync(DATA_DIR, { recursive: true });
writeFileSync(CONFIG_PATH, JSON.stringify(config, null, 2));
dataSync.trackWrite('agent-config.json');
}
// ── WS events ───────────────────────────────────────────────────────────────
export type { TurnStreamHooks };
export type WsEvent =
| { type: 'text_delta'; text: string }
| { type: 'tool_use'; tool: string; toolUseId: string; input: unknown }
| { type: 'tool_use_input'; toolUseId: string; input: unknown }
| { type: 'tool_result'; toolUseId: string; output: string; isError?: boolean }
| { type: 'tool_result_image'; toolUseId: string; dataUrl: string }
| { type: 'permission_request'; id: string; tool: string; input: Record<string, unknown> }
| { type: 'question_request'; id: string; questions: AskQuestion[] }
| { type: 'thinking_delta'; text: string }
| { type: 'done'; sessionId: string; stopReason?: 'end_turn' | 'max_turns_reached' | (string & {}); builtinHandled?: boolean }
| { type: 'model_resolved'; sessionId: string; model: string; engine: string; account?: ClaudeAccountRef }
| { type: 'error'; message: string }
| { type: 'stats'; sample: { t: number; cpu: number; mem: number; load: number; disk: number; diskUsedBytes?: number; diskTotalBytes?: number } };
// Add-on engines/features emit their OWN events (e.g. a duplex voice brain's `duplex_*`) through the
// object-typed `emitToSession()` bus (session-bus.ts) — NOT this union. So the core names none of them
// here, yet forwards them verbatim to clients. Keep this union the closed set of core-owned events.
// ── Stream chat ─────────────────────────────────────────────────────────────
const COMPACT_THRESHOLD = 30;
const RECENT_KEEP = 10;
export type PermissionHandler = (id: string, tool: string, input: Record<string, unknown>) => Promise<{ allow: boolean }>;
/** A single question the agent posed via the built-in AskUserQuestion tool. */
export type AskQuestion = {
question: string;
header: string;
multiSelect?: boolean;
options: { label: string; description: string; preview?: string }[];
};
/** Answers keyed by question text. String for single-select, string[] for multiSelect. */
export type QuestionAnswers = Record<string, string | string[]>;
/**
* Medium-agnostic handler for AskUserQuestion. Present the questions to a human
* (web UI, Slack, email, …), collect answers, and return them — they become the
* tool result. Return null if no human is reachable / dismissed / timed out, in
* which case the agent proceeds with its own best judgement.
*/
export type QuestionHandler = (id: string, questions: AskQuestion[]) => Promise<QuestionAnswers | null>;
function messagesSinceSummary(conv: ConvMessage[]): number {
const summaryIdx = conv.findLastIndex((m) =>
m.blocks.some((b) => b.type === 'summary')
);
return summaryIdx >= 0 ? conv.length - summaryIdx - 1 : conv.length;
}
function messagesToText(messages: ConvMessage[]): string {
return messages.map((m) => {
const role = m.role === 'user' ? 'User' : 'Assistant';
const texts = m.blocks
.filter((b) => b.type === 'text')
.map((b) => (b as { type: 'text'; text: string }).text)
.filter(Boolean);
return texts.length ? `${role}: ${texts.join('\n')}` : '';
}).filter(Boolean).join('\n\n');
}
async function maybeCompact(sessionId: string, conv: ConvMessage[]): Promise<void> {
const sinceLast = messagesSinceSummary(conv);
if (sinceLast < COMPACT_THRESHOLD) return;
const summaryIdx = conv.findLastIndex((m) =>
m.blocks.some((b) => b.type === 'summary')
);
const startIdx = summaryIdx >= 0 ? summaryIdx + 1 : 0;
const toCompact = conv.slice(startIdx, conv.length - RECENT_KEEP);
if (toCompact.length < RECENT_KEEP) return;
const existingSummary = summaryIdx >= 0
? (conv[summaryIdx].blocks.find((b) => b.type === 'summary') as { type: 'summary'; text: string }).text
: null;
const instruction = existingSummary
? `Here is an existing conversation summary:\n<existing_summary>\n${existingSummary}\n</existing_summary>\n\nHere are the new messages since that summary. Create an updated, comprehensive summary of the entire conversation so far. Be concise but preserve key facts, decisions, and context. Focus on what matters for continuing the conversation.`
: `Summarize this conversation concisely. Preserve key facts, decisions, open questions, and context needed to continue the conversation:`;
try {
const summaryText = await summarizeText(messagesToText(toCompact), instruction);
if (!summaryText) return;
const totalCompacted = (summaryIdx >= 0
? (conv[summaryIdx].blocks.find((b) => b.type === 'summary') as { type: 'summary'; text: string; compactedCount: number }).compactedCount
: 0) + toCompact.length;
const summaryMsg: ConvMessage = {
id: crypto.randomUUID(),
role: 'assistant',
blocks: [{ type: 'summary', text: summaryText, compactedCount: totalCompacted }],
};
const kept = conv.slice(conv.length - RECENT_KEEP);
const newConv = summaryIdx >= 0
? [...conv.slice(0, summaryIdx), summaryMsg, ...kept]
: [summaryMsg, ...kept];
saveConversation(sessionId, newConv);
conv.length = 0;
conv.push(...newConv);
console.log(`[claude] Compacted ${toCompact.length} messages for ${sessionId.slice(0, 8)} (total compacted: ${totalCompacted})`);
} catch (err) {
console.error('[claude] Compaction failed:', err);
}
}
function applyCompactMarkers(conv: ConvMessage[]): ConvMessage[] {
const markerIdx = conv.findLastIndex((m) =>
m.blocks.some((b) => b.type === 'compact_marker')
);
if (markerIdx < 0) return conv;
const marker = conv[markerIdx].blocks.find((b) => b.type === 'compact_marker') as { summary: string };
const summaryMsg: ConvMessage = {
id: 'compact-summary',
role: 'assistant',
blocks: [{ type: 'text', text: `<conversation_summary>\n${marker.summary}\n</conversation_summary>` }],
};
return [summaryMsg, ...conv.slice(markerIdx + 1)];
}
export interface AttachmentMeta { url: string; name: string; mimeType: string; path: string }
export interface StreamChatOpts {
/** Who this turn runs for. Required so every channel is enumerated by the compiler. Shadow mode:
* resolved + audited (turn.start/turn.end), not yet enforced. */
principal: Principal;
prompt: string;
attachments?: AttachmentMeta[];
images?: string[];
sessionId?: string;
uid: string;
userEmail?: string;
userName?: string;
mcpServers?: McpConfig;
abortController?: AbortController;
onPermissionRequest?: PermissionHandler;
onDestructiveApproval?: PermissionHandler;
onUserQuestion?: QuestionHandler;
/** See EngineStreamOpts.foregroundBashCommand. */
foregroundBashCommand?: string;
conversationReset?: boolean;
/** Opaque per-send bag from the client. Never interpreted here — handed to the turn-context seam,
* where an add-on's contributor reads its own keys. */
turnHints?: Record<string, unknown>;
context?: Record<string, string>;
/** Unattended scheduled run: `config.scheduleModel` (when set) replaces `config.model` as the default. */
scheduled?: boolean;
}
/** Run one agent turn. Security wrapper: resolves the principal's role/profile and audits turn.start / turn.end.
* SHADOW (SECURITY_ENFORCE off): the turn is unchanged; auditing can never alter or break it.
* ENFORCE: the session floor is lowered to the caller's rank (taint), the turn runs under a TurnGuard whose
* effective profile the engine applies, and it FAILS CLOSED — no runtime, a resolution error, or an effective
* profile with `outbound: false` (e.g. `none`) refuses the turn with an `error` event. */
export async function* streamChat(opts: StreamChatOpts): AsyncGenerator<WsEvent> {
const sec = security();
const enforce = enforcing();
if (!sec) {
if (!enforce) { yield* runTurn(opts); return; }
console.error('[security] SECURITY_ENFORCE is on but the security runtime is not initialized — refusing the turn');
yield { type: 'error', message: 'Security enforcement is on but not initialized, so this turn was not run.' };
return;
}
const { principal, sessionId } = opts;
let role: string | undefined;
let guard: TurnGuard | undefined;
let refusal: string | undefined;
try {
const d = sec.decide(principal, sessionId);
role = d.role;
let enforced: Record<string, unknown> = {};
if (enforce) {
// A broken policy fails closed at resolve time already; it must not ALSO permanently taint every session it touches.
if (sessionId && sec.policy.valid) lowerSessionFloor(sessionId, d.rank);
guard = new TurnGuard({ runtime: sec, principal, sessionId, role: d.role, rank: d.rank, floorOf: getSessionFloor });
const eff = guard.current();
if (!eff.profile.outbound) refusal = `This conversation runs as role "${eff.role}", which is not permitted to run the agent.`;
enforced = { enforced: true, effectiveRole: eff.role, effectiveProfile: eff.profileName, floor: sessionId ? getSessionFloor(sessionId) : undefined };
}
sec.record({ type: 'turn.start', principal: principal.id, role, sessionId, meta: { kind: principal.kind, lane: principal.attrs.lane, source: opts.context?.source, profile: d.profile, rank: d.rank, wouldDeny: d.wouldDeny, ...enforced } });
} catch (e: any) {
console.error('[security] turn.start audit failed:', e.message);
if (enforce) refusal = 'The security check for this turn failed, so it was not run.';
}
const started = Date.now();
let outcome = 'closed';
try {
if (enforce && (refusal || !guard)) {
outcome = 'denied';
console.warn(`[security] Refused turn for ${principal.id} session=${sessionId ?? 'new'}: ${refusal}`);
yield { type: 'error', message: refusal ?? 'The security check for this turn failed, so it was not run.' };
return;
}
for await (const ev of runTurn(opts, guard)) {
if (ev.type === 'done') outcome = 'done';
else if (ev.type === 'error') outcome = 'error';
yield ev;
}
if (outcome === 'closed') outcome = 'done';
} catch (e) {
outcome = 'threw';
throw e;
} finally {
try { sec.record({ type: 'turn.end', principal: principal.id, role, sessionId, reason: outcome, meta: { ms: Date.now() - started } }); }
catch (e: any) { console.error('[security] turn.end audit failed:', e.message); }
}
}
/** SECURITY_ENFORCE: `principal`'s rank under the live policy. Undefined in shadow mode, without a runtime, or on an
* invalid (fail-closed) policy — callers then apply no rank-based filtering. */
export function enforcedRank(principal: Principal): number | undefined {
const sec = security();
return enforcing() && sec?.policy.valid ? resolvePrincipal(sec.policy, principal).rank : undefined;
}
/** SECURITY_ENFORCE: input from `principal` entered `sessionId` outside a turn it runs (e.g. another human's Slack
* thread message) — lower the session floor to that principal's rank. No-op in shadow mode, without a runtime, or
* on an invalid (fail-closed) policy. Returns the new floor, or undefined when nothing was done. */
export async function taintSession(sessionId: string, principal: Principal | (() => Promise<Principal>)): Promise<number | undefined> {
const sec = security();
if (!enforcing() || !sec?.policy.valid) return undefined;
const p = typeof principal === 'function' ? await principal() : principal;
return lowerSessionFloor(sessionId, resolvePrincipal(sec.policy, p).rank);
}
/** The agent config a turn runs under. A scheduled run swaps in `scheduleModel` as its default model —
* at the config layer, so every engine's `directives.model || config.model` picks it up and an
* explicit `[model:…]` directive still wins. */
export function effectiveAgentConfig(opts: Pick<StreamChatOpts, 'scheduled'>): AgentConfig {
const config = getAgentConfig();
return opts.scheduled && config.scheduleModel ? { ...config, model: config.scheduleModel } : config;
}
async function* runTurn(opts: StreamChatOpts, guard?: TurnGuard): AsyncGenerator<WsEvent> {
const config = effectiveAgentConfig(opts);
const { prompt: cleanPrompt, directives: parsed, unresolvedModel } = parseDirectives(opts.prompt);
// An explicit model selection that resolves to nothing is a HARD stop, for the same reason an
// unregistered engine is: continuing would run the turn on `config.model` — a different model
// than the caller picked — with only a console.warn to show for it. Surfaced as a turn `error`
// event (not a throw) so WS, Slack, scheduler, MCP and webhook all report it identically and a
// scheduled run is marked failed instead of quietly succeeding on the wrong model.
if (unresolvedModel) {
const message = new ModelUnavailableError(unresolvedModel).message;
console.error(`[stream] ${message} (user=${opts.uid} session=${opts.sessionId ?? 'new'})`);
yield { type: 'error', message };
return;
}
const sessionMeta = opts.sessionId ? getSession(opts.sessionId) : undefined;
const directives: Directives = { ...sessionMeta?.directives, ...parsed };
// A two-model mode (delegate/advisor) is sticky like the model: `[off]` clears it, and a new mode
// replaces the other one. `modeOff` is a command, not a setting — never stored.
if (parsed.modeOff || parsed.advisor) delete directives.delegate;
if (parsed.modeOff || parsed.delegate) delete directives.advisor;
delete directives.modeOff;
// Persist ONLY what was explicitly chosen (inline directives, or the Agent Config panel's
// per-conversation save). Gaps stay gaps: every engine resolves `directives.x ?? config.x` at
// send time, so the global agent-config keeps applying live to sessions nobody pinned —
// including the non-UI channels (Slack, email, scheduler) that never open the config panel.
// Gap-filling here used to freeze the boot-time global onto every session's first turn, which
// made a later global engine/model change a no-op for every existing conversation.
if (opts.sessionId && JSON.stringify(directives) !== JSON.stringify(sessionMeta?.directives ?? {})) {
setSessionDirectives(opts.sessionId, directives);
console.log(`[claude] Directives: ${JSON.stringify(directives)}`);
yield { type: 'directives', directives } as any;
}
// Built-in /compact command
const slashCmd = parseSlashCommand(cleanPrompt);
if (slashCmd?.command === 'compact') {
console.log('[claude] /compact intercepted — handling as built-in');
const sessionId = opts.sessionId;
if (!sessionId) {
yield { type: 'text_delta', text: 'Nothing to compact — no active session.' };
yield { type: 'done', sessionId: '', builtinHandled: true } as any;
return;
}
const conv = loadConversation(sessionId);
const lastMarkerIdx = conv.findLastIndex((m) => m.blocks.some((b) => b.type === 'compact_marker'));
const toSummarize = conv.slice(lastMarkerIdx >= 0 ? lastMarkerIdx + 1 : 0);
if (toSummarize.length < 4) {
yield { type: 'text_delta', text: 'Conversation too short to compact.' };
yield { type: 'done', sessionId, builtinHandled: true } as any;
return;
}
try {
const existingSummary = lastMarkerIdx >= 0
? (conv[lastMarkerIdx].blocks.find((b) => b.type === 'compact_marker') as { summary: string }).summary
: null;
const instruction = existingSummary
? `Here is an existing conversation summary:\n<existing_summary>\n${existingSummary}\n</existing_summary>\n\nHere are the new messages since that summary. Create an updated, comprehensive summary of the entire conversation so far. Be concise but preserve key facts, decisions, and context. Output only the summary — no preamble, no questions, no conversational filler.`
: `Summarize this conversation concisely. Preserve key facts, decisions, open questions, and context needed to continue the conversation. Output only the summary — no preamble, no questions, no conversational filler.`;
const summaryText = await summarizeText(messagesToText(toSummarize), instruction);
if (!summaryText) {
yield { type: 'text_delta', text: 'Failed to generate summary.' };
yield { type: 'done', sessionId, builtinHandled: true } as any;
return;
}
const prevCompacted = lastMarkerIdx >= 0
? (conv[lastMarkerIdx].blocks.find((b) => b.type === 'compact_marker') as { compactedCount: number }).compactedCount
: 0;
const compactedCount = prevCompacted + toSummarize.length;
appendMessage(sessionId, {
id: crypto.randomUUID(),
role: 'assistant',
blocks: [{ type: 'compact_marker', summary: summaryText, compactedCount }],
});
yield { type: 'compact_marker', summary: summaryText, compactedCount, sessionId } as any;
} catch (err) {
console.error('[claude] /compact failed:', err);
yield { type: 'text_delta', text: 'Compaction failed.' };
}
yield { type: 'done', sessionId, builtinHandled: true } as any;
return;
}
// Slash command resolution
let effectivePrompt = cleanPrompt;
if (slashCmd) {
const skill = getSkill(slashCmd.command);
if (skill) {
const { body } = parseSkillFrontmatter(skill.content);
effectivePrompt = formatCommandBlock(slashCmd.command, body, slashCmd.args);
if (skill.meta.model) directives.model = skill.meta.model;
} else if (opts.mcpServers && slashCmd.command in opts.mcpServers) {
effectivePrompt = getMcpCommandPrompt(slashCmd.command, slashCmd.args);
} else {
yield { type: 'text_delta', text: `Unknown command: /${slashCmd.command}` };
yield { type: 'done', sessionId: opts.sessionId ?? '' };
return;
}
}
// Expand @skill + @workspace-file mentions
const withSkillMentions = expandMentionedSkills(effectivePrompt);
const withMentions = expandWorkspaceMentions(withSkillMentions);
const defaultSkills = resolveDefaultSkillsContent();
const mcpSkills = opts.mcpServers ? buildMcpSkillHintsBlock(Object.keys(opts.mcpServers)) : '';
const discoveryEnabled = config.skillDiscovery !== false;
const skillIndex = discoveryEnabled ? buildSkillIndexBlock() : '';
// Triggered skills are sticky: once matched in a session, they stay injected on every later turn.
const stickyNames = opts.sessionId ? getSession(opts.sessionId)?.triggeredSkills ?? [] : [];
const newTriggerNames = discoveryEnabled ? matchTriggeredSkillNames(effectivePrompt, opts.context) : [];
const triggeredNames = [...new Set([...stickyNames, ...newTriggerNames])];
// Per-name blocks so an engine that resumes across turns can send only newly triggered ones. Joined
// exactly as skillInjectionBlocks(triggeredNames) joins them, so contextBlock is byte-identical.
const triggeredSkillBlocks = triggeredNames.map((n) => [n, skillInjectionBlocks([n])] as const).filter(([, b]) => b);
const triggeredSkills = triggeredSkillBlocks.map(([, b]) => b).join('\n');
// Per-skill turn budget. Resolved HERE, once every invoked skill is known (the slash command and
// the triggered/sticky set), and GAP-FILLING only: an inline `[turns:N]` or a session-pinned
// value already sits in `directives` and is left alone. The global `config.maxTurns` is applied
// later still, by the engine (`directives.turns ?? config.maxTurns`), so a skill budget slots
// cleanly between the two without touching the global default.
// NOTE the deliberate asymmetry with `model` above, which OVERWRITES an inline directive: a
// long-running skill needs a floor, not the last word — the human who typed `[turns:5]` to probe
// it cheaply must still get 5.
if (!directives.turns) {
const skillTurns = resolveSkillTurns([slashCmd?.command, ...triggeredNames]);
if (skillTurns) {
directives.turns = skillTurns;
console.log(`[claude] Directives: ${JSON.stringify(directives)} (turns from skill frontmatter)`);
}
}
const workspaceTree = buildWorkspaceContextBlock();
const offloadBlock = buildOffloadContextBlock();
const contact = opts.userEmail ? contacts.find({ email: opts.userEmail }) : null;
const userBlock = contacts.formatUserBlock(contact);
const teamRoster = contacts.formatRoster();
const userContextBlock = getUserContextBlock(contact);
const contextBlock = [userBlock, userContextBlock, teamRoster, defaultSkills, triggeredSkills, skillIndex, mcpSkills, workspaceTree, offloadBlock].filter(Boolean).join('\n');
const contextSections: Record<string, string> = Object.fromEntries(Object.entries({
user: userBlock, userContext: userContextBlock, roster: teamRoster, defaultSkills,
...Object.fromEntries(triggeredSkillBlocks.map(([n, b]) => [`skill:${n}`, b])),
skillIndex, mcpSkills, workspace: workspaceTree, offload: offloadBlock,
}).filter(([, v]) => v));
// Load conversation for the engine
const sessionId = opts.sessionId ?? crypto.randomUUID();
if (newTriggerNames.length) {
// Some channels (WS) upsert the session only after streaming — ensure the record exists first.
// email may be absent on this path (userEmail is optional); keep the value as-is —
// `as string` asserts the record contract without altering the stored value.
upsertSession(sessionId, effectivePrompt, { uid: opts.uid, email: opts.userEmail as string, name: opts.userName });
addTriggeredSkills(sessionId, newTriggerNames);
}
let conversation: ConvMessage[] = [];
if (opts.sessionId) {
conversation = loadConversation(opts.sessionId);
if (conversation.length > 0) await maybeCompact(opts.sessionId, conversation);
conversation = applyCompactMarkers(conversation);
}
// Per-turn add-on context: prepend whatever the turn-context seam's contributors return to THIS
// turn's prompt only (not the cacheable contextBlock, not the persisted user message). The core
// registers no contributors, so this is a no-op here and the prompt is unchanged.
let turnPrompt = withMentions;
const turnContext = collectTurnContext({ sessionId, uid: opts.uid, hints: opts.turnHints });
if (turnContext) {
turnPrompt = `${turnContext}\n\n${withMentions}`;
console.log(`[stream] turn-context injected (${turnContext.length} chars) for session=${sessionId}`);
}
// Resolve engine and delegate. An unregistered engine is a HARD stop: rerouting to claude-code
// would switch provider and billing under the caller while the UI still showed the requested one.
// Surfaced as a turn `error` event (not a throw) so every transport — WS, Slack, scheduler, MCP,
// webhook — reports it the same way and a scheduled run is marked failed instead of dying.
let engine: ReturnType<typeof resolveAndGetEngine>;
try {
engine = resolveAndGetEngine(directives as any, config);
} catch (err) {
const message = (err as Error).message;
console.error(`[stream] ${message} (user=${opts.uid} session=${sessionId})`);
yield { type: 'error', message };
return;
}
// An engine that doesn't apply the profile can only run a turn that needs no restriction.
if (guard && !engine.enforcesProfile) {
const { profile, role } = guard.current();
if (!(profile.tools.includes('*') && profile.mcps.includes('*') && profile.env.includes('*'))) {
const message = `Engine "${engine.name}" does not enforce security profiles, so it cannot run this conversation's role "${role}".`;
console.error(`[security] ${message} (user=${opts.uid} session=${sessionId})`);
yield { type: 'error', message };
return;
}
}
console.log(`[stream] engine=${engine.name} user=${opts.uid} session=${sessionId}`);
// Another engine's turn never reaches the claude-code transcript, so a stored SDK-resume mapping is
// stale from here on — mark it; the claude-code engine then falls back to a fresh query (engine-switch).
const resumeState = getSession(sessionId)?.claudeResume;
if (engine.name !== 'claude-code' && resumeState && !resumeState.interruptedBy) setClaudeResume(sessionId, undefined, { interruptedBy: engine.name });
yield* engine.stream({
prompt: turnPrompt,
conversation,
contextBlock,
contextSections,
attachments: opts.attachments,
images: opts.images,
sessionId,
uid: opts.uid,
userEmail: opts.userEmail,
userName: opts.userName,
mcpServers: opts.mcpServers,
abortController: opts.abortController,
onPermissionRequest: opts.onPermissionRequest,
onDestructiveApproval: opts.onDestructiveApproval,
onUserQuestion: opts.onUserQuestion,
foregroundBashCommand: opts.foregroundBashCommand,
turnHints: opts.turnHints,
conversationReset: opts.conversationReset,
context: opts.context,
...(guard ? { security: guard } : {}),
directives,
config,
});
}
/** The notice appended to a turn that ran out of steps. Shared so every transport says the same
* thing — the MCP path said NOTHING at all, so a truncated turn arrived looking finished. */
export const MAX_TURNS_NOTICE = '\n\n---\n⚠️ Reached the maximum number of steps for this turn. Send "continue" to pick up where I left off.';
export async function consumeStream(
stream: AsyncGenerator<WsEvent>,
onEvent?: (ev: WsEvent) => void,
hooks?: TurnStreamHooks,
): Promise<ConvBlock[]> {
const acc = createTurnAccumulator(hooks ?? {});
for await (const ev of stream) {
onEvent?.(ev);
if (acc.push(ev)) break;
}
return acc.finish();
}