diff --git a/scripts/source-file-size-baseline.json b/scripts/source-file-size-baseline.json index 0552b03f31..055867f3c9 100644 --- a/scripts/source-file-size-baseline.json +++ b/scripts/source-file-size-baseline.json @@ -46,6 +46,6 @@ "src/websocket/listener/file-commands.ts": 1053, "src/websocket/listener/lifecycle.ts": 1051, "src/websocket/listener/protocol-inbound.ts": 2264, - "src/websocket/listener/protocol-outbound.ts": 1085, + "src/websocket/listener/protocol-outbound.ts": 1083, "src/websocket/listener/turn.ts": 1010 } diff --git a/src/runtime-context.ts b/src/runtime-context.ts index e49854a8ec..55e7b4f620 100644 --- a/src/runtime-context.ts +++ b/src/runtime-context.ts @@ -19,6 +19,8 @@ export interface RuntimeContextSnapshot { conversationId?: string | null; skillsDirectory?: string | null; skillSources?: SkillSource[]; + /** Runtime override for whether tools may project the agent's MemFS. */ + memfsEnabled?: boolean; workingDirectory?: string | null; /** * Set when the runtime-scoped working directory was found deleted and diff --git a/src/tools/impl/shell-env.ts b/src/tools/impl/shell-env.ts index ca49787198..7c8394f387 100644 --- a/src/tools/impl/shell-env.ts +++ b/src/tools/impl/shell-env.ts @@ -375,10 +375,11 @@ export function getShellEnv(): NodeJS.ProcessEnv { const localBackendEnabled = process.env.LETTA_LOCAL_BACKEND_EXPERIMENTAL === "1" || process.env.LETTA_LOCAL_BACKEND_EXPERIMENTAL?.toLowerCase() === "true"; - if ( - !localBackendNoMemfs && - (settingsManager.isMemfsEnabled(agentId) || localBackendEnabled) - ) { + const runtimeMemfsEnabled = getRuntimeContext()?.memfsEnabled; + const memfsEnabled = + runtimeMemfsEnabled ?? + (settingsManager.isMemfsEnabled(agentId) || localBackendEnabled); + if (!localBackendNoMemfs && memfsEnabled) { const memoryDir = resolveScopedMemoryDir({ agentId }); if (!memoryDir) { throw new Error("Unable to resolve memory directory"); diff --git a/src/tools/toolset.ts b/src/tools/toolset.ts index a6cae98a83..6eafb8c489 100644 --- a/src/tools/toolset.ts +++ b/src/tools/toolset.ts @@ -375,6 +375,7 @@ export async function prepareToolExecutionContextForScope(params: { modContext?: ModContext; modEvents?: ModEvents; modAdapters?: ModAdapter[]; + runtimeContext?: Partial; }): Promise { const { connectionId, @@ -396,6 +397,7 @@ export async function prepareToolExecutionContextForScope(params: { modContext, modEvents, modAdapters, + runtimeContext, } = params; const backend = getBackend(); @@ -462,6 +464,7 @@ export async function prepareToolExecutionContextForScope(params: { modAdapters, agent: agent as AgentState, runtimeContext: { + ...runtimeContext, connectionId, environmentDeviceId, agentId, diff --git a/src/types/protocol_v2.ts b/src/types/protocol_v2.ts index 28faaa3ed3..cd7ee51eba 100644 --- a/src/types/protocol_v2.ts +++ b/src/types/protocol_v2.ts @@ -783,13 +783,11 @@ export interface RuntimeStartCreateConversationOptions { /** Body forwarded to the Letta conversations create API. */ body?: Omit; } - export interface RuntimeStartClientInfo { name: string; title?: string; version?: string; } - export interface RuntimeStartCommand { type: "runtime_start"; /** Echoed back in the response for request correlation. */ @@ -808,6 +806,8 @@ export interface RuntimeStartCommand { cwd?: string | null; /** Initial permission mode for this runtime scope. */ mode?: DevicePermissionMode; + /** Runtime-only policy that skips local MemFS reads and writes. */ + stateless?: boolean; skill_sources?: readonly ("bundled" | "global" | "agent" | "project")[]; /** Preserve the current override when skill_sources is omitted. */ preserve_skill_sources?: boolean; /** Optional client metadata for diagnostics/future protocol negotiation. */ diff --git a/src/websocket/listener/commands.ts b/src/websocket/listener/commands.ts index 1655d32480..2efe6fb442 100644 --- a/src/websocket/listener/commands.ts +++ b/src/websocket/listener/commands.ts @@ -6,7 +6,6 @@ import { applySetMaxContext, formatSetMaxContextResult, } from "@/agent/max-context"; -import { getScopedMemoryFilesystemRoot } from "@/agent/memory-filesystem"; import { REMEMBER_PROMPT } from "@/agent/prompt-assets"; import type { ConversationMessageCompactBody } from "@/backend"; import { getBackend } from "@/backend"; @@ -54,6 +53,10 @@ import { } from "./protocol-outbound"; import { flushRemoteSettingsWrites } from "./remote-settings"; import { clearConversationRuntimeState, emitListenerStatus } from "./runtime"; +import { + getConversationMemoryDirectory, + isConversationMemfsEnabled, +} from "./runtime-memory"; import { ensureSecretsHydratedForAgent, invalidateSecretsCacheForAgent, @@ -111,6 +114,15 @@ export async function handleExecuteCommand( try { let output: string; + if ( + conversationRuntime.stateless && + ["doctor", "init", "remember", "reflect"].includes(command.command_id) + ) { + throw new Error( + `/${command.command_id} is unavailable in a stateless runtime because it writes agent memory.`, + ); + } + switch (command.command_id) { case "clear": output = await handleClearCommand(socket, conversationRuntime, { @@ -341,7 +353,10 @@ export async function handleReloadCommand( ); } - await reloadListenerModAdapter(listener, conversationRuntime.agentId); + await reloadListenerModAdapter( + listener, + conversationRuntime.stateless ? undefined : conversationRuntime.agentId, + ); if (conversationRuntime.agentId) { invalidateSecretsCacheForAgent(listener, conversationRuntime.agentId); @@ -558,7 +573,7 @@ async function handleCompactCommand( ); if ( reflectionSettings.trigger === "compaction-event" && - settingsManager.isMemfsEnabled(agentId) + isConversationMemfsEnabled(conversationRuntime) ) { void buildMaybeLaunchReflectionSubagent({ runtime: conversationRuntime, @@ -687,9 +702,8 @@ async function handleDoctorCommand( } const { context: gitContext } = gatherInitGitContext(); - const memoryDir = settingsManager.isMemfsEnabled(agentId) - ? getScopedMemoryFilesystemRoot(agentId) - : undefined; + const memoryDir = + getConversationMemoryDirectory(conversationRuntime) ?? undefined; const skillNameFrontmatterRepair = await repairMissingSkillNameFrontmatter(memoryDir); const skillNameFrontmatterRepairReport = @@ -747,9 +761,8 @@ async function handleInitCommand( } const { context: gitContext } = gatherInitGitContext(); - const memoryDir = settingsManager.isMemfsEnabled(agentId) - ? getScopedMemoryFilesystemRoot(agentId) - : undefined; + const memoryDir = + getConversationMemoryDirectory(conversationRuntime) ?? undefined; const initMessage = buildInitMessage({ gitContext, memoryDir }); @@ -900,7 +913,7 @@ async function handleReflectCommand( const result = await launchReflectionSubagent({ agentId, conversationId, - memfsEnabled: settingsManager.isMemfsEnabled(agentId), + memfsEnabled: isConversationMemfsEnabled(conversationRuntime), triggerSource: "manual", description: "Reflecting on conversation", recompileByConversation: listener.systemPromptRecompileByConversation, diff --git a/src/websocket/listener/commands/runtime-start-skill-sources.test.ts b/src/websocket/listener/commands/runtime-start-skill-sources.test.ts index d104d2bc93..85adf7c34b 100644 --- a/src/websocket/listener/commands/runtime-start-skill-sources.test.ts +++ b/src/websocket/listener/commands/runtime-start-skill-sources.test.ts @@ -62,6 +62,68 @@ describe("runtime_start skill sources", () => { } }); + test("keeps stateless policy across idle runtime eviction until reset", async () => { + const storageDir = await mkdtemp(join(tmpdir(), "runtime-stateless-")); + try { + const backend = new LocalBackend({ + storageDir, + executionMode: "deterministic", + }); + __testSetBackend(backend); + const agent = await backend.createAgent({ + name: "Stateless SDK worker", + model: "anthropic/claude-sonnet-4-6", + } as AgentCreateBody); + const listener = createRuntime(); + const context = { + socket: {} as WebSocket, + connectionId: "test-connection", + runtime: listener, + safeSocketSend: () => true, + runDetachedListenerTask: () => {}, + getOrCreateScopedRuntime, + replaySyncStateForRuntime: async () => {}, + }; + + await handleRuntimeStartCommand( + { + type: "runtime_start", + request_id: "runtime-stateless", + agent_id: agent.id, + conversation_id: "default", + stateless: true, + recover_approvals: false, + }, + context, + ); + + const scoped = getOrCreateScopedRuntime(listener, agent.id, "default"); + expect(scoped.stateless).toBe(true); + expect(listener.statelessByConversation?.has(scoped.key)).toBe(true); + expect(evictConversationRuntimeIfIdle(scoped)).toBe(true); + expect( + getOrCreateScopedRuntime(listener, agent.id, "default").stateless, + ).toBe(true); + + await handleRuntimeStartCommand( + { + type: "runtime_start", + request_id: "runtime-stateful", + agent_id: agent.id, + conversation_id: "default", + recover_approvals: false, + }, + context, + ); + expect( + getOrCreateScopedRuntime(listener, agent.id, "default").stateless, + ).toBe(false); + expect(listener.statelessByConversation?.has(scoped.key)).toBe(false); + } finally { + await rm(storageDir, { recursive: true, force: true }); + } + }); + test("preserves skill sources when a secondary controller updates runtime tools", async () => { const storageDir = await mkdtemp(join(tmpdir(), "runtime-skills-")); try { diff --git a/src/websocket/listener/commands/runtime-start.ts b/src/websocket/listener/commands/runtime-start.ts index c6e01ebe48..f3eb5a0c3a 100644 --- a/src/websocket/listener/commands/runtime-start.ts +++ b/src/websocket/listener/commands/runtime-start.ts @@ -22,6 +22,7 @@ import { persistPermissionModeMapForRuntime, } from "@/websocket/listener/permission-mode"; import { isRuntimeStartCommand } from "@/websocket/listener/protocol-inbound"; +import { setConversationRuntimeStateless } from "@/websocket/listener/runtime-memory"; import type { ConversationRuntime, ListenerConnectionId, @@ -305,6 +306,8 @@ async function applyRuntimeStartState( ); } + setConversationRuntimeStateless(scopedRuntime, parsed.stateless === true); + if (parsed.mode) { const mode = migratePermissionMode(parsed.mode); if (!mode) { diff --git a/src/websocket/listener/lifecycle.ts b/src/websocket/listener/lifecycle.ts index 8ee8d3dd2e..a89d45f818 100644 --- a/src/websocket/listener/lifecycle.ts +++ b/src/websocket/listener/lifecycle.ts @@ -284,7 +284,6 @@ function stampInboundUserMessageOtids( } export function createRuntime(): ListenerRuntime { - const bootWorkingDirectory = getCurrentWorkingDirectory(); return { socket: null, transport: null, @@ -312,11 +311,12 @@ export function createRuntime(): ListenerRuntime { pendingQueueEmitScope: undefined, onWsEvent: undefined, reminderState: createSharedReminderState(), - bootWorkingDirectory, + bootWorkingDirectory: getCurrentWorkingDirectory(), workingDirectoryByConversation: loadPersistedCwdMap(), worktreeWatcherByConversation: new Map(), permissionModeByConversation: loadPersistedPermissionModeMap(), skillSourcesByConversation: new Map(), + statelessByConversation: new Set(), reminderStateByConversation: new Map(), contextTrackerByConversation: new Map(), systemPromptRecompileByConversation: new Map(), diff --git a/src/websocket/listener/message-router.ts b/src/websocket/listener/message-router.ts index c82184a639..54c5980728 100644 --- a/src/websocket/listener/message-router.ts +++ b/src/websocket/listener/message-router.ts @@ -4,7 +4,6 @@ import { estimateSystemPromptTokensFromMemoryDir, setSystemPromptDoctorState, } from "@/cli/helpers/system-prompt-warning"; -import { settingsManager } from "@/settings-manager"; import type { AbortMessageCommand, ApprovalResponseBody, @@ -63,6 +62,7 @@ import { } from "./queue"; import { emitLoopErrorNotice } from "./recoverable-notices"; import { getActiveRuntime, safeEmitWsEvent } from "./runtime"; +import { isConversationMemfsEnabled } from "./runtime-memory"; import type { ListenerTransport } from "./transport"; import { handleIncomingMessage } from "./turn"; import type { @@ -761,7 +761,12 @@ export function createListenerMessageHandler( // Internal-only: refresh doctor state after recompile (no chat output) if (parsed.command_id === "refresh_doctor_state") { const agentId = parsed.runtime.agent_id; - if (agentId && settingsManager.isMemfsEnabled(agentId)) { + const doctorRuntime = getOrCreateScopedRuntime( + runtime, + agentId, + parsed.runtime.conversation_id, + ); + if (agentId && isConversationMemfsEnabled(doctorRuntime)) { try { const { getScopedMemoryFilesystemRoot } = await import( "@/agent/memory-filesystem" diff --git a/src/websocket/listener/mod-adapter.ts b/src/websocket/listener/mod-adapter.ts index 736761d956..5231ee3ffb 100644 --- a/src/websocket/listener/mod-adapter.ts +++ b/src/websocket/listener/mod-adapter.ts @@ -91,6 +91,7 @@ export function createListenerModContext( reasoning_effort?: string | null; } | null; } | null; + memfsEnabled?: boolean; modelIdentifier?: string | null; permissionMode?: string | null; toolset?: string | null; @@ -107,12 +108,21 @@ export function createListenerModContext( }); return { ...context, - memfs: resolveListenerAgentMemfsContext(context.agent.id), + memfs: + options.memfsEnabled === false + ? { enabled: false, memoryDir: null } + : resolveListenerAgentMemfsContext(context.agent.id), }; } -export function createListenerAgentModContext(agentId: string): ModContext { - return createListenerModContext({ agent: { id: agentId } }); +export function createListenerAgentModContext( + agentId: string, + memfsEnabled?: boolean, +): ModContext { + return createListenerModContext({ + agent: { id: agentId }, + ...(memfsEnabled !== undefined ? { memfsEnabled } : {}), + }); } export function createListenerModAdapter( @@ -235,8 +245,12 @@ export async function ensureListenerAgentModAdapter( export async function ensureListenerModAdaptersForAgent( runtime: ListenerRuntime, agentId: string, + options: { includeAgent?: boolean } = {}, ): Promise { const globalAdapter = ensureListenerModAdapter(runtime); + if (options.includeAgent === false) { + return [globalAdapter]; + } const agentAdapter = await ensureListenerAgentModAdapter(runtime, agentId); return agentAdapter ? [globalAdapter, agentAdapter] : [globalAdapter]; } diff --git a/src/websocket/listener/protocol-inbound.test.ts b/src/websocket/listener/protocol-inbound.test.ts index 34e739b966..1c823e0341 100644 --- a/src/websocket/listener/protocol-inbound.test.ts +++ b/src/websocket/listener/protocol-inbound.test.ts @@ -109,6 +109,7 @@ describe("agent/conversation management protocol-inbound validators", () => { conversation_source_tags: ["channel:slack"], cwd: "/tmp/project", mode: "acceptEdits", + stateless: true, skill_sources: [], preserve_skill_sources: true, client_info: { name: "test", title: "Test", version: "1.0.0" }, @@ -239,6 +240,12 @@ describe("agent/conversation management protocol-inbound validators", () => { agent_id: "agent-1", mode: "bad", }, + { + type: "runtime_start", + request_id: "r0", + agent_id: "agent-1", + stateless: "yes", + }, { type: "runtime_start", request_id: "r0", diff --git a/src/websocket/listener/protocol-inbound.ts b/src/websocket/listener/protocol-inbound.ts index 5d19731535..305a205009 100644 --- a/src/websocket/listener/protocol-inbound.ts +++ b/src/websocket/listener/protocol-inbound.ts @@ -522,7 +522,6 @@ function isRuntimeStartClientInfo(value: unknown): boolean { (value.version === undefined || typeof value.version === "string") ); } - export function isRuntimeStartCommand( value: unknown, ): value is RuntimeStartCommand { @@ -542,6 +541,7 @@ export function isRuntimeStartCommand( isStringArray(c.conversation_source_tags)) && (c.cwd === undefined || c.cwd === null || typeof c.cwd === "string") && (c.mode === undefined || isDevicePermissionMode(c.mode)) && + (c.stateless === undefined || typeof c.stateless === "boolean") && (c.skill_sources === undefined || isSkillSourceArray(c.skill_sources)) && (c.preserve_skill_sources === undefined || typeof c.preserve_skill_sources === "boolean") && diff --git a/src/websocket/listener/protocol-outbound.ts b/src/websocket/listener/protocol-outbound.ts index b08521cbc7..1eac3e83d4 100644 --- a/src/websocket/listener/protocol-outbound.ts +++ b/src/websocket/listener/protocol-outbound.ts @@ -1,6 +1,5 @@ import type { MessageCreate } from "@letta-ai/letta-client/resources/agents/agents"; import type { LettaStreamingResponse } from "@letta-ai/letta-client/resources/agents/messages"; -import { getScopedMemoryFilesystemRoot } from "@/agent/memory-filesystem"; import { getSubagents } from "@/agent/subagent-state"; import { getGitContext } from "@/cli/helpers/git-context"; import { getReflectionSettings } from "@/cli/helpers/memory-reminder"; @@ -53,6 +52,7 @@ import { hasInterruptedCacheForScope, safeEmitWsEvent, } from "./runtime"; +import { runtimeMemoryDir } from "./runtime-memory"; import { resolveRuntimeScope, resolveScopedAgentId, @@ -267,9 +267,7 @@ export function buildDeviceStatus( ? [] : getPendingControlRequests(listener, scope), experiments: experimentManager.list(), - memory_directory: scopedAgentId - ? getScopedMemoryFilesystemRoot(scopedAgentId) - : null, + memory_directory: runtimeMemoryDir(conversationRuntime, scopedAgentId), ...(params === undefined ? { cwd_map: getExportedCwdMap(listener), diff --git a/src/websocket/listener/runtime-memory.test.ts b/src/websocket/listener/runtime-memory.test.ts new file mode 100644 index 0000000000..d73c5101cd --- /dev/null +++ b/src/websocket/listener/runtime-memory.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, test } from "bun:test"; +import { + getConversationSkillSources, + setConversationRuntimeStateless, +} from "./runtime-memory"; +import type { + ConversationRuntime, + ListenerConnectionState, + ListenerRuntime, +} from "./types"; + +function makeRuntime(): { + runtime: ConversationRuntime; + connection: ListenerConnectionState; +} { + const connection = {} as ListenerConnectionState; + const listener = { + connections: new Map([["connection-1", connection]]), + statelessByConversation: new Set(), + } as ListenerRuntime; + const runtime = { + key: "agent:agent-1::conversation:conv-1", + listener, + agentId: "agent-1", + stateless: false, + skillSources: undefined, + } as ConversationRuntime; + return { runtime, connection }; +} + +describe("stateless conversation runtime policy", () => { + test("keeps the policy on the conversation runtime scope", () => { + const { runtime } = makeRuntime(); + + setConversationRuntimeStateless(runtime, true); + + expect(runtime.stateless).toBe(true); + expect(runtime.listener.statelessByConversation?.has(runtime.key)).toBe( + true, + ); + + setConversationRuntimeStateless(runtime, false); + expect(runtime.stateless).toBe(false); + expect(runtime.listener.statelessByConversation?.has(runtime.key)).toBe( + false, + ); + }); + + test("removes agent-scoped skills without changing explicit global sources", () => { + const { runtime } = makeRuntime(); + runtime.stateless = true; + runtime.skillSources = ["bundled", "agent", "project"]; + + expect(getConversationSkillSources(runtime)).toEqual([ + "bundled", + "project", + ]); + }); + + test("uses non-agent skill defaults for stateless sessions", () => { + const { runtime } = makeRuntime(); + runtime.stateless = true; + + expect(getConversationSkillSources(runtime)).toEqual([ + "bundled", + "global", + "project", + ]); + }); +}); diff --git a/src/websocket/listener/runtime-memory.ts b/src/websocket/listener/runtime-memory.ts new file mode 100644 index 0000000000..3506f7c3c4 --- /dev/null +++ b/src/websocket/listener/runtime-memory.ts @@ -0,0 +1,65 @@ +import { getScopedMemoryFilesystemRoot } from "@/agent/memory-filesystem"; +import type { SkillSource } from "@/agent/skill-sources"; +import { settingsManager } from "@/settings-manager"; +import type { ConversationRuntime } from "./types"; + +const STATELESS_DEFAULT_SKILL_SOURCES: SkillSource[] = [ + "bundled", + "global", + "project", +]; + +export function setConversationRuntimeStateless( + runtime: ConversationRuntime, + stateless: boolean, +): void { + runtime.stateless = stateless; + let statelessScopes = runtime.listener.statelessByConversation; + if (!statelessScopes) { + statelessScopes = new Set(); + runtime.listener.statelessByConversation = statelessScopes; + } + if (stateless) { + statelessScopes.add(runtime.key); + } else { + statelessScopes.delete(runtime.key); + } +} + +export function isConversationMemfsEnabled( + runtime: ConversationRuntime, +): boolean { + return ( + !runtime.stateless && + runtime.agentId !== null && + settingsManager.isMemfsEnabled(runtime.agentId) + ); +} + +export function getConversationMemoryDirectory( + runtime: ConversationRuntime, +): string | null { + return isConversationMemfsEnabled(runtime) && runtime.agentId + ? getScopedMemoryFilesystemRoot(runtime.agentId) + : null; +} + +export function runtimeMemoryDir( + runtime: ConversationRuntime | null, + agentId: string | null, +): string | null { + return agentId && runtime?.stateless !== true + ? getScopedMemoryFilesystemRoot(agentId) + : null; +} + +export function getConversationSkillSources( + runtime: ConversationRuntime, +): SkillSource[] | undefined { + if (!runtime.stateless) { + return runtime.skillSources; + } + return (runtime.skillSources ?? STATELESS_DEFAULT_SKILL_SOURCES).filter( + (source) => source !== "agent", + ); +} diff --git a/src/websocket/listener/runtime.ts b/src/websocket/listener/runtime.ts index 6138c6048b..80edc409b0 100644 --- a/src/websocket/listener/runtime.ts +++ b/src/websocket/listener/runtime.ts @@ -253,6 +253,7 @@ export function createConversationRuntime( agentId: normalizedAgentId, conversationId: normalizedConversationId, skillSources: listener.skillSourcesByConversation.get(runtimeKey)?.slice(), + stateless: listener.statelessByConversation?.has(runtimeKey) === true, activeConnectionId: null, turnLifecycle, messageQueue: Promise.resolve(), diff --git a/src/websocket/listener/turn-cleanup.ts b/src/websocket/listener/turn-cleanup.ts index c89eb9b404..f2bed0b0f7 100644 --- a/src/websocket/listener/turn-cleanup.ts +++ b/src/websocket/listener/turn-cleanup.ts @@ -1,11 +1,11 @@ import { runPostTurnMemorySync } from "@/reminders/memory-git-sync"; import { enqueueMemoryGitSyncReminder } from "@/reminders/state"; -import { settingsManager } from "@/settings-manager"; import { persistPermissionModeMapForRuntime, pruneConversationPermissionModeStateIfDefault, } from "./permission-mode"; import { emitDeviceStatusIfOpen } from "./protocol-outbound"; +import { isConversationMemfsEnabled } from "./runtime-memory"; import type { ConversationRuntime } from "./types"; export async function runListenerTurnCleanup(params: { @@ -30,7 +30,7 @@ export async function runListenerTurnCleanup(params: { if (agentId) { await runPostTurnMemorySync({ agentId, - isEnabled: (id) => settingsManager.isMemfsEnabled(id), + isEnabled: () => isConversationMemfsEnabled(runtime), debugLabel: "Post-turn listener memory sync", enqueueReminder: (text) => { enqueueMemoryGitSyncReminder(runtime.reminderState, { text }); diff --git a/src/websocket/listener/turn-completion.ts b/src/websocket/listener/turn-completion.ts index 3cae9b274a..16637db60b 100644 --- a/src/websocket/listener/turn-completion.ts +++ b/src/websocket/listener/turn-completion.ts @@ -4,8 +4,8 @@ import type { Line } from "@/cli/helpers/accumulator"; import { getReflectionSettings } from "@/cli/helpers/memory-reminder"; import { maybeLaunchPostTurnReflection } from "@/cli/helpers/post-turn-reflection"; import { appendTranscriptDeltaJsonl } from "@/cli/helpers/reflection-transcript"; -import { settingsManager } from "@/settings-manager"; import { debugWarn } from "@/utils/debug"; +import { isConversationMemfsEnabled } from "./runtime-memory"; import type { ListenerTransport } from "./transport"; import { buildMaybeLaunchReflectionSubagent, @@ -35,6 +35,7 @@ export async function completeSuccessfulListenerTurn(params: { workingDirectory: params.workingDirectory, permissionMode: params.permissionMode, cachedAgent: params.getCachedAgent(), + stateless: params.runtime.stateless, }); if (params.isInterrupted()) { return "interrupted"; @@ -55,7 +56,10 @@ export async function completeSuccessfulListenerTurn(params: { } try { - if (params.transcriptLines.length > 0) { + if ( + isConversationMemfsEnabled(params.runtime) && + params.transcriptLines.length > 0 + ) { await appendTranscriptDeltaJsonl( params.agentId, params.conversationId, @@ -82,7 +86,7 @@ export async function completeSuccessfulListenerTurn(params: { await maybeLaunchPostTurnReflection({ agentId: params.agentId, conversationId: params.conversationId, - memfsEnabled: settingsManager.isMemfsEnabled(params.agentId), + memfsEnabled: isConversationMemfsEnabled(params.runtime), reflectionSettings, reminderState: params.runtime.reminderState, contextTracker: params.runtime.contextTracker, diff --git a/src/websocket/listener/turn-events.ts b/src/websocket/listener/turn-events.ts index bbda697044..8e99f06e08 100644 --- a/src/websocket/listener/turn-events.ts +++ b/src/websocket/listener/turn-events.ts @@ -12,7 +12,6 @@ import { launchReflectionSubagent, } from "@/cli/helpers/reflection-launcher"; import { getTurnStartCancel } from "@/mods/turn-start-cancel"; -import { settingsManager } from "@/settings-manager"; import { getListenerTelemetrySurface } from "@/telemetry"; import type { StreamDelta } from "@/types/protocol_v2"; import { @@ -21,6 +20,7 @@ import { ensureListenerModAdaptersForAgent, } from "./mod-adapter"; import { emitCanonicalMessageDelta } from "./protocol-outbound"; +import { isConversationMemfsEnabled } from "./runtime-memory"; import type { ListenerTransport } from "./transport"; import type { ConversationRuntime, ListenerRuntime } from "./types"; @@ -56,17 +56,20 @@ export async function emitListenerTurnStart(options: { workingDirectory: string; permissionMode?: string | null; cachedAgent?: AgentState | null; + stateless?: boolean; }): Promise { try { const modAdapters = await ensureListenerModAdaptersForAgent( options.runtime, options.agentId, + { includeAgent: !options.stateless }, ); const context = createListenerModContext({ sessionId: options.conversationId, workingDirectory: options.workingDirectory, permissionMode: options.permissionMode ?? null, agent: options.cachedAgent ?? { id: options.agentId }, + ...(options.stateless ? { memfsEnabled: false } : {}), }); const event = { agentId: options.agentId, @@ -102,17 +105,20 @@ export async function emitListenerTurnEnd(options: { workingDirectory: string; permissionMode?: string | null; cachedAgent?: AgentState | null; + stateless?: boolean; }): Promise { try { const modAdapters = await ensureListenerModAdaptersForAgent( options.runtime, options.agentId, + { includeAgent: !options.stateless }, ); const context = createListenerModContext({ sessionId: options.conversationId, workingDirectory: options.workingDirectory, permissionMode: options.permissionMode ?? null, agent: options.cachedAgent ?? { id: options.agentId }, + ...(options.stateless ? { memfsEnabled: false } : {}), }); const event: { agentId: string; @@ -154,7 +160,7 @@ export function buildMaybeLaunchReflectionSubagent(params: { const result = await launchReflectionSubagent({ agentId, conversationId, - memfsEnabled: settingsManager.isMemfsEnabled(agentId), + memfsEnabled: isConversationMemfsEnabled(runtime), triggerSource, reflectionSettings, description: AUTO_REFLECTION_DESCRIPTION, diff --git a/src/websocket/listener/turn-setup.ts b/src/websocket/listener/turn-setup.ts index 5e46dfa6f2..05da693777 100644 --- a/src/websocket/listener/turn-setup.ts +++ b/src/websocket/listener/turn-setup.ts @@ -28,6 +28,10 @@ import { ensureListenerModAdaptersForAgent, } from "./mod-adapter"; import type { ConversationPermissionModeState } from "./permission-mode"; +import { + getConversationSkillSources, + isConversationMemfsEnabled, +} from "./runtime-memory"; import { emitListenerTurnStart } from "./turn-events"; import { createTurnInputState, @@ -92,7 +96,7 @@ export async function prepareListenerTurn(params: { let listenAgentMetadata = await ensureListenerWarmStateForTurn( runtime.listener, - { agentId, conversationId }, + { agentId, conversationId, stateless: runtime.stateless }, ); if (isInterrupted()) { return { kind: "interrupted" }; @@ -138,19 +142,21 @@ export async function prepareListenerTurn(params: { cachedAgent = (await getBackend().retrieveAgent(agentId, { include: ["agent.tags"], })) as AgentState; - const { - ensureLettaCodeOriginTag, - getMemoryPromptModeForAgent, - scheduleManagedSystemPromptUpdate, - } = await import("@/agent/system-prompt-versioning"); - cachedAgent = await ensureLettaCodeOriginTag(cachedAgent); - scheduleManagedSystemPromptUpdate({ - agent: cachedAgent, - memoryMode: getMemoryPromptModeForAgent(cachedAgent.id), - onUpdated: (updatedAgent) => { - cachedAgent = updatedAgent; - }, - }); + if (!runtime.stateless) { + const { + ensureLettaCodeOriginTag, + getMemoryPromptModeForAgent, + scheduleManagedSystemPromptUpdate, + } = await import("@/agent/system-prompt-versioning"); + cachedAgent = await ensureLettaCodeOriginTag(cachedAgent); + scheduleManagedSystemPromptUpdate({ + agent: cachedAgent, + memoryMode: getMemoryPromptModeForAgent(cachedAgent.id), + onUpdated: (updatedAgent) => { + cachedAgent = updatedAgent; + }, + }); + } } catch (error) { debugWarn( "listen", @@ -233,6 +239,7 @@ export async function prepareListenerTurn(params: { workingDirectory, permissionMode: permissionModeState.mode, cachedAgent, + stateless: runtime.stateless, }) : ({ cancelled: false, handlerCount: 0, input: messagesToSend } as const); if (isInterrupted()) { @@ -267,6 +274,7 @@ export async function prepareListenerTurn(params: { const modAdapters = await ensureListenerModAdaptersForAgent( runtime.listener, agentId, + { includeAgent: !runtime.stateless }, ); const listenerOptions = connectionId ? runtime.listener.connections.get(connectionId)?.options @@ -288,11 +296,17 @@ export async function prepareListenerTurn(params: { workingDirectory, permissionModeState, skillsDirectory: listenerOptions?.skillsDirectory, - skillSources: runtime.skillSources, + skillSources: getConversationSkillSources(runtime), cachedAgent, - modContext: createListenerAgentModContext(agentId), + modContext: createListenerAgentModContext( + agentId, + isConversationMemfsEnabled(runtime), + ), modAdapters, modEvents: createListenerModEvents(modAdapters), + runtimeContext: { + memfsEnabled: isConversationMemfsEnabled(runtime), + }, }); if (isInterrupted()) { return { kind: "interrupted" }; diff --git a/src/websocket/listener/types.ts b/src/websocket/listener/types.ts index c20900371d..0d415120bf 100644 --- a/src/websocket/listener/types.ts +++ b/src/websocket/listener/types.ts @@ -190,6 +190,8 @@ export type ConversationRuntime = { conversationId: string; /** Runtime-scoped SDK override. Undefined uses the process defaults. */ skillSources: SkillSource[] | undefined; + /** Skip local MemFS reads and writes for this conversation runtime. */ + stateless: boolean; /** Connection currently executing this conversation's turn, if client-owned. */ activeConnectionId: ListenerConnectionId | null; turnLifecycle: TurnLifecycle; @@ -341,6 +343,8 @@ export type ListenerRuntime = { >; /** Per-conversation skill overrides survive idle ConversationRuntime eviction. */ skillSourcesByConversation: Map; + /** Stateless runtime scopes survive idle ConversationRuntime eviction. */ + statelessByConversation?: Set; /** Per-conversation reminder state survives ConversationRuntime eviction. */ reminderStateByConversation: Map; /** Per-conversation context tracker survives ConversationRuntime eviction. */ diff --git a/src/websocket/listener/warmup.ts b/src/websocket/listener/warmup.ts index ee630687fd..abdf8f6d57 100644 --- a/src/websocket/listener/warmup.ts +++ b/src/websocket/listener/warmup.ts @@ -3,6 +3,7 @@ import { debugWarn } from "@/utils/debug"; import { ensureMemfsSyncedForAgent } from "./memfs-sync"; import { ensureListenerAgentModAdapter } from "./mod-adapter"; import { emitDeviceStatusUpdateIfChanged } from "./protocol-outbound"; +import { getConversationRuntime } from "./runtime"; import { ensureSecretsHydratedForAgent } from "./secrets-sync"; import { isListenerTransportOpen } from "./transport"; import type { ListenerRuntime } from "./types"; @@ -16,6 +17,7 @@ export type ListenerAgentMetadata = { export type ListenerWarmupScope = { agentId: string; conversationId: string; + stateless?: boolean; }; function getAgentMetadataPromise( @@ -91,14 +93,15 @@ export async function ensureListenerWarmStateForTurn( try { await Promise.all([ - warmupDeps.ensureMemfsSyncedForAgent(listener, agentId), + ...(scope.stateless + ? [] + : [warmupDeps.ensureMemfsSyncedForAgent(listener, agentId)]), warmupDeps.ensureSecretsHydratedForAgent(listener, agentId), agentMetadataPromise, ]); - const agentModAdapter = await ensureListenerAgentModAdapter( - listener, - agentId, - ); + const agentModAdapter = scope.stateless + ? null + : await ensureListenerAgentModAdapter(listener, agentId); const transport = listener.transport ?? listener.socket; if (agentModAdapter && transport && isListenerTransportOpen(transport)) { emitDeviceStatusUpdateIfChanged(transport, listener, { @@ -134,9 +137,16 @@ export function scheduleListenerWarmupsAfterSync( return; } + const normalizedConversationId = conversationId ?? "default"; + const conversationRuntime = getConversationRuntime( + listener, + agentId, + normalizedConversationId, + ); void ensureListenerWarmStateForTurn(listener, { agentId, - conversationId: conversationId ?? "default", + conversationId: normalizedConversationId, + stateless: conversationRuntime?.stateless === true, }).catch((error) => { debugWarn( "listener-warmup",