diff --git a/src/agents/embedded-agent-runner/compact.hooks.test.ts b/src/agents/embedded-agent-runner/compact.hooks.test.ts index a5f410976b3..ef96b2664fe 100644 --- a/src/agents/embedded-agent-runner/compact.hooks.test.ts +++ b/src/agents/embedded-agent-runner/compact.hooks.test.ts @@ -5,8 +5,9 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { expectDefined } from "@openclaw/normalization-core"; import type { AgentMessage } from "openclaw/plugin-sdk/agent-core"; -import { beforeAll, beforeEach, describe, expect, it, vi, type Mock } from "vitest"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi, type Mock } from "vitest"; import { createDeferred } from "../../../test/helpers/promise.js"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; import { createReplyOperation } from "../../auto-reply/reply/reply-run-registry.js"; import { upsertSessionEntryCore } from "../../config/sessions/session-accessor.js"; import type { PluginManifestRecord } from "../../plugins/manifest-registry.js"; @@ -78,6 +79,8 @@ let compactTesting: typeof import("./compact.js").testing; let onSessionTranscriptUpdate: typeof import("../../sessions/transcript-events.js").onSessionTranscriptUpdate; let onInternalSessionTranscriptUpdate: typeof import("../../sessions/transcript-events.js").onInternalSessionTranscriptUpdate; +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + const TEST_SESSION_ID = "session-1"; const TEST_SESSION_KEY = "agent:main:session-1"; const TEST_SESSION_FILE = "/tmp/session.jsonl"; @@ -2793,7 +2796,9 @@ describe("compactEmbeddedAgentSession hooks (ownsCompaction engine)", () => { expect(result.result).not.toHaveProperty("summary"); }); - it("binds context-engine compaction runtime LLM to the session agent", async () => { + it("fails closed for a fallback-owned legacy compaction target", async () => { + const legacySessionId = "legacy-session-47"; + const legacyStorePath = join(tempDirs.make("openclaw-legacy-compaction-"), "openclaw.sqlite"); resolveSessionAgentIdsMock.mockReturnValueOnce({ defaultAgentId: "main", sessionAgentId: "lossless-agent", @@ -2808,7 +2813,14 @@ describe("compactEmbeddedAgentSession hooks (ownsCompaction engine)", () => { }, }, }, + sessionId: legacySessionId, sessionKey: "legacy-topic-47", + sessionTarget: { + agentId: "lossless-agent", + sessionId: legacySessionId, + sessionKey: "legacy-topic-47", + storePath: legacyStorePath, + }, }), ); @@ -2835,6 +2847,46 @@ describe("compactEmbeddedAgentSession hooks (ownsCompaction engine)", () => { await expect( runtimeContext.llm?.complete?.({ messages: [{ role: "user", content: "summarize" }], + }), + ).rejects.toThrow("not bound to an active session agent"); + }); + + it("binds a queued legacy compaction from its explicit owner field", async () => { + const legacySessionId = "explicit-legacy-session-48"; + await compactEmbeddedAgentSession( + wrappedCompactionArgs({ + config: { agents: { defaults: { model: "openai/gpt-5.5" } } }, + contextEngineAgentId: "lossless-agent", + sessionId: legacySessionId, + sessionKey: "legacy-topic-48", + sessionTarget: { + agentId: "lossless-agent", + sessionId: legacySessionId, + sessionKey: "legacy-topic-48", + storePath: join(tempDirs.make("openclaw-explicit-legacy-compaction-"), "openclaw.sqlite"), + }, + }), + ); + + const compactInput = ( + contextEngineCompactMock.mock.calls as unknown as Array< + [ + { + runtimeContext?: { + llm?: { + complete?: (params: { + messages: Array<{ role: "user"; content: string }>; + agentId?: string; + }) => Promise; + }; + }; + }, + ] + > + )[0]?.[0]; + await expect( + compactInput?.runtimeContext?.llm?.complete?.({ + messages: [{ role: "user", content: "summarize" }], agentId: "other-agent", }), ).rejects.toThrow("cannot override the active session agent"); diff --git a/src/agents/embedded-agent-runner/compact.queued.ts b/src/agents/embedded-agent-runner/compact.queued.ts index aa1dd1a5753..2b877dae1f7 100644 --- a/src/agents/embedded-agent-runner/compact.queued.ts +++ b/src/agents/embedded-agent-runner/compact.queued.ts @@ -1,6 +1,7 @@ /** * Queues embedded-agent session compaction onto the correct command lane. */ +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../context-engine/host-compat.js"; import { ensureContextEnginesInitialized } from "../../context-engine/init.js"; import { @@ -158,6 +159,7 @@ async function disposeContextEngine(contextEngine: ContextEngine): Promise async function deferOwningContextEngineBudgetCompaction(params: { compactParams: CompactEmbeddedAgentSessionParams; + contextEngineSessionKey?: string; contextEngine: ContextEngine; contextEngineRuntimeContext: ContextEngineRuntimeContext; contextEngineRuntimeSettings: ContextEngineRuntimeSettings; @@ -168,13 +170,14 @@ async function deferOwningContextEngineBudgetCompaction(params: { await runContextEngineMaintenance({ contextEngine: params.contextEngine, sessionId: params.compactParams.sessionId, - sessionKey: params.compactParams.sessionKey, + sessionKey: params.contextEngineSessionKey ?? params.compactParams.sessionKey, sessionTarget: buildContextEngineCompactionSessionTarget(params.compactParams), sessionFile: params.compactParams.sessionFile, reason: "turn", runtimeContext: params.contextEngineRuntimeContext, runtimeSettings: params.contextEngineRuntimeSettings, config: params.compactParams.config, + contextEngineAgentId: params.compactParams.contextEngineAgentId, disposeDeferredContextEngineAfterMaintenance: true, onDeferredMaintenance: () => { deferredScheduled = true; @@ -243,6 +246,11 @@ function mergeSecondaryNativeHarnessCompactionDetails(params: { export async function compactEmbeddedAgentSession( params: CompactEmbeddedAgentSessionParams, ): Promise { + const contextEngineAgentId = + normalizeOptionalString(params.contextEngineAgentId) ?? normalizeOptionalString(params.agentId); + const contextEngineSessionKey = + normalizeOptionalString(params.sessionKey) ?? + normalizeOptionalString(params.sessionTarget?.sessionKey); const runtimeTarget = await resolveAgentRunSessionTarget({ ...params, missingSessionKey: "resolve-existing", @@ -254,9 +262,10 @@ export async function compactEmbeddedAgentSession( sessionKey: runtimeTarget.sessionKey, sessionTarget: runtimeTarget, sessionFile: runtimeTarget.sessionKey, + contextEngineAgentId, }; if (resolvedParams.trigger !== "manual") { - return await compactEmbeddedAgentSessionImpl(resolvedParams); + return await compactEmbeddedAgentSessionImpl(resolvedParams, contextEngineSessionKey); } // Reply operations and embedded handles are separate lifecycle owners. A // /compact reply may coexist with this handle, but another embedded writer may not. @@ -289,10 +298,13 @@ export async function compactEmbeddedAgentSession( resolvedParams.sessionFile, ); try { - return await compactEmbeddedAgentSessionImpl({ - ...resolvedParams, - abortSignal, - }); + return await compactEmbeddedAgentSessionImpl( + { + ...resolvedParams, + abortSignal, + }, + contextEngineSessionKey, + ); } finally { clearActiveEmbeddedRun( resolvedParams.sessionId, @@ -305,6 +317,7 @@ export async function compactEmbeddedAgentSession( async function compactEmbeddedAgentSessionImpl( inputParams: CompactEmbeddedAgentSessionParams, + contextEngineSessionKey?: string, ): Promise { if (inputParams.abortSignal?.aborted) { return createCompactionAbortedResult(); @@ -399,6 +412,7 @@ async function compactEmbeddedAgentSessionImpl( agentDir, resolvedWorkspaceDir, lease.snapshot, + contextEngineSessionKey, () => { disposeContextEngineOnExit = false; }, @@ -422,6 +436,7 @@ async function compactResolvedContextEngine( agentDir: string, resolvedWorkspaceDir: string, preparedModelRuntime: PreparedModelRuntimeSnapshot, + contextEngineSessionKey: string | undefined, releaseContextEngineOwnership: () => void, ): Promise { const runtimeTarget = await resolveAgentRunSessionTarget({ @@ -576,6 +591,7 @@ async function compactResolvedContextEngine( params: preparedParams, agentDir, harnessRuntime: preparedHarnessRuntime, + contextEngineSessionKey, contextTokenBudget, contextEnginePluginId: resolveContextEngineOwnerPluginId(contextEngine), }); @@ -620,6 +636,7 @@ async function compactResolvedContextEngine( ) { const deferredResult = await deferOwningContextEngineBudgetCompaction({ compactParams: preparedParams, + contextEngineSessionKey, contextEngine, contextEngineRuntimeContext, contextEngineRuntimeSettings, @@ -768,7 +785,7 @@ async function compactResolvedContextEngine( await runContextEngineMaintenance({ contextEngine, sessionId: postCompactionSessionId, - sessionKey: params.sessionKey, + sessionKey: contextEngineSessionKey ?? params.sessionKey, sessionTarget: buildContextEngineCompactionSessionTarget({ ...params, sessionFile: postCompactionSessionFile, @@ -780,6 +797,7 @@ async function compactResolvedContextEngine( runtimeContext, runtimeSettings: contextEngineRuntimeSettings, config: params.config, + contextEngineAgentId: params.contextEngineAgentId, }); } if (engineOwnsCompaction && result.ok && result.compacted) { @@ -922,16 +940,12 @@ function shouldAttemptNativeHarnessCompaction(params: { function buildCompactionContextEngineRuntimeContext(params: { params: CompactEmbeddedAgentSessionParams; agentDir: string; + contextEngineSessionKey?: string; harnessRuntime?: string; contextEnginePluginId?: string; contextTokenBudget?: number; }): ContextEngineRuntimeContext { - const { sessionAgentId } = resolveSessionAgentIds({ - sessionKey: params.params.sessionKey, - config: params.params.config, - agentId: params.params.agentId, - }); - const { sessionFile: _sessionFile, ...runtimeParams } = params.params; + const { sessionFile: _sessionFile, contextEngineAgentId, ...runtimeParams } = params.params; return { ...runtimeParams, sessionTarget: buildContextEngineCompactionSessionTarget(params.params), @@ -943,8 +957,8 @@ function buildCompactionContextEngineRuntimeContext(params: { }), ...resolveContextEngineCapabilities({ config: params.params.config, - sessionKey: params.params.sessionKey, - agentId: sessionAgentId, + sessionKey: params.contextEngineSessionKey ?? params.params.sessionKey, + explicitAgentId: contextEngineAgentId, authProfileId: params.params.authProfileId, contextEnginePluginId: params.contextEnginePluginId, purpose: "context-engine.compaction", diff --git a/src/agents/embedded-agent-runner/compact.types.ts b/src/agents/embedded-agent-runner/compact.types.ts index 8f752479988..9aefdec6328 100644 --- a/src/agents/embedded-agent-runner/compact.types.ts +++ b/src/agents/embedded-agent-runner/compact.types.ts @@ -21,6 +21,8 @@ import type { ScheduledToolPolicyContext } from "../scheduled-tool-policy.js"; import type { TrustedSubagentCompletionHandoff } from "../subagents/announce/subagent-announce-handoff.js"; export type CompactEmbeddedAgentSessionParams = { + /** Explicit session owner captured before fallback agent resolution. */ + contextEngineAgentId?: string; sessionId: string; runId?: string; sessionKey?: string; diff --git a/src/agents/embedded-agent-runner/context-engine-capabilities.ts b/src/agents/embedded-agent-runner/context-engine-capabilities.ts index 9f4d34488c6..ef7cb2d7f83 100644 --- a/src/agents/embedded-agent-runner/context-engine-capabilities.ts +++ b/src/agents/embedded-agent-runner/context-engine-capabilities.ts @@ -9,7 +9,7 @@ import { resolveBoundAgentIdForSession } from "../session-agent-binding.js"; type ResolveContextEngineCapabilitiesParams = { config?: OpenClawConfig; sessionKey?: string; - agentId?: string; + explicitAgentId?: string; authProfileId?: string; contextEnginePluginId?: string; purpose: string; @@ -25,7 +25,7 @@ export function resolveContextEngineCapabilities( const agentId = resolveBoundAgentIdForSession({ config: params.config, sessionKey, - agentId: params.agentId, + agentId: params.explicitAgentId, }); const contextEnginePluginId = normalizeOptionalString(params.contextEnginePluginId); return { diff --git a/src/agents/embedded-agent-runner/context-engine-maintenance.ts b/src/agents/embedded-agent-runner/context-engine-maintenance.ts index 018ca3cff0b..3359c1bbf32 100644 --- a/src/agents/embedded-agent-runner/context-engine-maintenance.ts +++ b/src/agents/embedded-agent-runner/context-engine-maintenance.ts @@ -59,6 +59,7 @@ type ContextEngineMaintenanceParams = { runtimeContext?: ContextEngineRuntimeContext; runtimeSettings?: ContextEngineRuntimeSettings; agentId?: string; + contextEngineAgentId?: string; executionMode?: "foreground" | "background"; onDeferredMaintenance?: (promise: Promise) => void; onDeferredMaintenanceFailure?: (error: unknown) => void; @@ -236,7 +237,7 @@ function buildContextEngineMaintenanceRuntimeContext( ...resolveContextEngineCapabilities({ config: params.config, sessionKey: params.sessionKey, - agentId: params.agentId, + explicitAgentId: params.contextEngineAgentId, authProfileId: normalizeOptionalString(params.runtimeContext?.authProfileId), contextEnginePluginId: params.contextEnginePluginId, purpose: params.purpose ?? "context-engine.maintenance", diff --git a/src/agents/embedded-agent-runner/run-orchestrator.ts b/src/agents/embedded-agent-runner/run-orchestrator.ts index b783097b9d1..888d03c890a 100644 --- a/src/agents/embedded-agent-runner/run-orchestrator.ts +++ b/src/agents/embedded-agent-runner/run-orchestrator.ts @@ -102,6 +102,9 @@ export function runEmbeddedAgent( async function runEmbeddedAgentInternal( paramsInput: RunEmbeddedAgentInternalParams, ): Promise { + const contextEngineAgentId = + normalizeOptionalString(paramsInput.sessionTarget?.agentId) ?? + normalizeOptionalString(paramsInput.agentId); const paramsBase = applyAgentRunSessionTargetIdentity(paramsInput); const skillWorkshopProposalMutationBudget = paramsBase.skillWorkshopProposalOnly ? (paramsBase.skillWorkshopProposalMutationBudget ?? { remaining: 1 }) @@ -412,6 +415,7 @@ async function runEmbeddedAgentInternal( return await executePreparedEmbeddedRun({ runParams: params, + contextEngineAgentId, provider, modelId, agentDir, diff --git a/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts b/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts index 8d2749d8cd0..c1c93b25203 100644 --- a/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts +++ b/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts @@ -1,20 +1,28 @@ import path from "node:path"; -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../context-engine/host-compat.js"; import { buildContextEngineRuntimeSettings } from "../../context-engine/runtime-settings.js"; -import type { ContextEngine } from "../../context-engine/types.js"; +import type { ContextEngine, ContextEngineRuntimeContext } from "../../context-engine/types.js"; import { createTestAdmittedRunContext } from "../admitted-run-context.test-support.js"; import type { AgentRuntimeAuthPlan } from "../runtime-plan/types.js"; import { compactEmbeddedRunForRecovery, createEmbeddedRunCompactionRuntime, + type EmbeddedRunCompactionRecoveryInput, } from "./run/compaction-runtime.js"; import { createEmbeddedRunContextRecoveryState } from "./run/context-recovery-state.js"; import type { PreparedEmbeddedRunInput } from "./run/execution-context.js"; import type { EmbeddedRunAttemptResult } from "./run/types.js"; const tempDirs = useAutoCleanupTempDirTracker(afterEach); +const completionMocks = vi.hoisted(() => ({ + prepareSimpleCompletionModelForAgent: vi.fn(), + completeWithPreparedSimpleCompletionModel: vi.fn(), + resolveSimpleCompletionSelectionForAgent: vi.fn(), +})); + +vi.mock("../simple-completion-runtime.js", () => completionMocks); // Keep this dedicated leaf on the compaction composition boundary. Runtime/auth/lane policy is // covered at its direct owners so this shard never reloads the complete public runner graph. @@ -58,7 +66,84 @@ function makeContextEngine(compact = vi.fn()): ContextEngine { } as ContextEngine; } +function makeRecoveryInput( + overrides: Partial = {}, +): EmbeddedRunCompactionRecoveryInput { + const runParams = overrides.runParams ?? baseRunParams; + return { + runParams, + state: createEmbeddedRunContextRecoveryState(), + contextEngine: makeContextEngine(), + genericCompactionRecoveryAllowed: true, + attempt: makeAttempt(), + runtimeAuthPlan: { + providerForAuth: "openai", + authProfileProviderForAuth: "openai", + }, + resolvedSessionKey: runParams.sessionKey ?? baseRunParams.sessionKey, + sessionAgentId: "main", + agentDir: "/tmp/agent", + workspaceDir: "/tmp/workspace", + provider: "openai", + modelId: "gpt-5.5", + harnessRuntime: "openclaw", + thinkLevel: "off", + authProfileIdSource: "auto", + resolveContextEnginePluginId: () => undefined, + buildRuntimeSettings: ({ tokenBudget, degradedReason }) => + buildContextEngineRuntimeSettings({ + contextEngineHost: OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST, + provider: "openai", + requestedModel: "gpt-5.5", + resolvedModel: "gpt-5.5", + promptTokenBudget: tokenBudget, + degradedReason, + }), + onCompactionHookMessages: vi.fn(async () => {}), + runOwnsCompactionBeforeHook: vi.fn(async () => {}), + runOwnsCompactionAfterHook: vi.fn(async () => {}), + adoptCompactionTranscript: vi.fn(async () => undefined), + getActiveSession: () => ({ + id: "session-1", + file: runParams.sessionFile ?? runParams.sessionKey ?? runParams.sessionId, + }), + prepareCompactedTranscriptRetry: vi.fn(async () => {}), + armPostCompactionGuard: vi.fn(), + ...overrides, + }; +} + describe("compactEmbeddedRunForRecovery", () => { + beforeEach(() => { + completionMocks.prepareSimpleCompletionModelForAgent.mockReset(); + completionMocks.completeWithPreparedSimpleCompletionModel.mockReset(); + completionMocks.resolveSimpleCompletionSelectionForAgent.mockReset(); + completionMocks.prepareSimpleCompletionModelForAgent.mockResolvedValue({ + selection: { provider: "openai", modelId: "gpt-5.5", agentDir: "/tmp/main" }, + model: { + provider: "openai", + id: "gpt-5.5", + name: "gpt-5.5", + api: "openai", + input: ["text"], + reasoning: false, + contextWindow: 128_000, + maxTokens: 4096, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + }, + auth: { apiKey: "test-api-key", source: "test", mode: "api-key" }, + }); + completionMocks.completeWithPreparedSimpleCompletionModel.mockResolvedValue({ + content: [{ type: "text", text: "done" }], + usage: { input: 1, output: 1, total: 2 }, + }); + completionMocks.resolveSimpleCompletionSelectionForAgent.mockReturnValue({ + provider: "openai", + modelId: "gpt-5.5", + agentDir: "/tmp/main", + }); + }); + it("carries locked model, auth, fallback, cache, and overflow facts into compaction", async () => { const compact = vi.fn(async () => ({ ok: true as const, @@ -78,46 +163,20 @@ describe("compactEmbeddedRunForRecovery", () => { } satisfies AgentRuntimeAuthPlan; const result = await compactEmbeddedRunForRecovery( - { + makeRecoveryInput({ runParams: { ...baseRunParams, modelSelectionLocked: true, modelFallbacksOverride: [], }, - state: createEmbeddedRunContextRecoveryState(), contextEngine, contextTokenBudget: 200_000, - genericCompactionRecoveryAllowed: true, attempt: makeAttempt({ promptCache }), runtimeAuthPlan, - resolvedSessionKey: baseRunParams.sessionKey, - sessionAgentId: "main", - agentDir: "/tmp/agent", - workspaceDir: "/tmp/workspace", - provider: "openai", - modelId: "gpt-5.5", - harnessRuntime: "openclaw", thinkLevel: "ultra", authProfileId: "openai:work", authProfileIdSource: "user", - resolveContextEnginePluginId: () => undefined, - buildRuntimeSettings: ({ tokenBudget, degradedReason }) => - buildContextEngineRuntimeSettings({ - contextEngineHost: OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST, - provider: "openai", - requestedModel: "gpt-5.5", - resolvedModel: "gpt-5.5", - promptTokenBudget: tokenBudget, - degradedReason, - }), - onCompactionHookMessages: vi.fn(async () => {}), - runOwnsCompactionBeforeHook: vi.fn(async () => {}), - runOwnsCompactionAfterHook: vi.fn(async () => {}), - adoptCompactionTranscript: vi.fn(async () => undefined), - getActiveSession: () => ({ id: "session-1", file: baseRunParams.sessionFile }), - prepareCompactedTranscriptRetry: vi.fn(async () => {}), - armPostCompactionGuard: vi.fn(), - }, + }), { tokenBudget: 200_000, trigger: "overflow", @@ -150,6 +209,40 @@ describe("compactEmbeddedRunForRecovery", () => { }, }); }); + + it("does not trust the active run fallback during recovery compaction", async () => { + const compact = vi.fn(async (params: { runtimeContext?: ContextEngineRuntimeContext }) => { + await params.runtimeContext?.llm?.complete({ + messages: [{ role: "user", content: "summarize" }], + }); + return { ok: true as const, compacted: false as const }; + }); + const contextEngine = makeContextEngine(compact); + const runParams = { + ...baseRunParams, + config: { agents: { defaults: { model: "openai/gpt-5.5" } } }, + sessionKey: "legacy-session", + sessionFile: "legacy-session", + } satisfies PreparedEmbeddedRunInput["runParams"]; + + await expect( + compactEmbeddedRunForRecovery( + makeRecoveryInput({ + runParams, + contextEngine, + resolvedSessionKey: "legacy-session", + }), + { + tokenBudget: 200_000, + trigger: "overflow", + diagId: "diag-unbound", + attempt: 1, + maxAttempts: 3, + }, + ), + ).rejects.toThrow("not bound to an active session agent"); + expect(completionMocks.prepareSimpleCompletionModelForAgent).not.toHaveBeenCalled(); + }); }); describe("createEmbeddedRunCompactionRuntime", () => { diff --git a/src/agents/embedded-agent-runner/run/attempt-dispatch-preparation.ts b/src/agents/embedded-agent-runner/run/attempt-dispatch-preparation.ts index 6adbe33f16d..ca23400c570 100644 --- a/src/agents/embedded-agent-runner/run/attempt-dispatch-preparation.ts +++ b/src/agents/embedded-agent-runner/run/attempt-dispatch-preparation.ts @@ -206,6 +206,7 @@ export async function prepareAndDispatchEmbeddedRunAttempt(input: { ? { kind: "caller-owned", sessionManager: params.sessionManager } : { kind: "runtime-target", sessionTarget: resolvedSessionTarget }, runtime: { + contextEngineAgentId: runInput.contextEngineAgentId, sessionId: sessionPromptState.sessionId, sessionFile: sessionPromptState.sessionFile, sessionKey: resolvedSessionKey, diff --git a/src/agents/embedded-agent-runner/run/attempt-finalize.ts b/src/agents/embedded-agent-runner/run/attempt-finalize.ts index def1cbdb266..c60d78699cb 100644 --- a/src/agents/embedded-agent-runner/run/attempt-finalize.ts +++ b/src/agents/embedded-agent-runner/run/attempt-finalize.ts @@ -278,6 +278,7 @@ export async function completeEmbeddedAttemptAfterTurn( withSessionManagerRewriteLock: transcript.withSessionManagerRewriteLock, config: attempt.config, agentId: runtime.sessionAgentId, + contextEngineAgentId: attempt.contextEngineAgentId, }), sessionManager: transcript.sessionManager, config: attempt.config, diff --git a/src/agents/embedded-agent-runner/run/attempt-prompt-helpers.ts b/src/agents/embedded-agent-runner/run/attempt-prompt-helpers.ts index 3513f439f9c..d97b4be88d2 100644 --- a/src/agents/embedded-agent-runner/run/attempt-prompt-helpers.ts +++ b/src/agents/embedded-agent-runner/run/attempt-prompt-helpers.ts @@ -493,6 +493,7 @@ export function resolveAttemptMediaTaskSystemPromptAddition(params: { type AfterTurnRuntimeContextAttempt = Pick< EmbeddedRunAttemptParams, | "sessionTarget" + | "contextEngineAgentId" | "sessionKey" | "sandboxSessionKey" | "messageChannel" @@ -601,7 +602,7 @@ export function buildAfterTurnRuntimeContext(params: { ...resolveContextEngineCapabilities({ config: params.attempt.config, sessionKey: params.attempt.sessionKey, - agentId: params.activeAgentId, + explicitAgentId: params.attempt.contextEngineAgentId, authProfileId: params.attempt.authProfileId, contextEnginePluginId: params.contextEnginePluginId, purpose: "context-engine.after-turn", diff --git a/src/agents/embedded-agent-runner/run/attempt-recovery.ts b/src/agents/embedded-agent-runner/run/attempt-recovery.ts index 5d403070a64..6a3dcf17976 100644 --- a/src/agents/embedded-agent-runner/run/attempt-recovery.ts +++ b/src/agents/embedded-agent-runner/run/attempt-recovery.ts @@ -220,6 +220,7 @@ export async function recoverEmbeddedRunAttempt(input: { runtimeAuthPlan: runtimePlan.auth, resolvedSessionKey: runInput.resolvedSessionKey, sessionAgentId: input.sessionAgentId, + contextEngineAgentId: runInput.contextEngineAgentId, agentDir: runInput.agentDir, workspaceDir: runInput.workspaceDir, provider: compactionSelection.provider, diff --git a/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts index 92aa9e1214f..cfc6b7bc4a3 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts @@ -521,6 +521,7 @@ export async function prepareEmbeddedAttemptSessionManager(input: { runtimeSettings: contextParams.runtimeSettings, config: attempt.config, agentId: input.sessionAgentId, + contextEngineAgentId: attempt.contextEngineAgentId, }), warn: (message) => log.warn(message), }); diff --git a/src/agents/embedded-agent-runner/run/compaction-runtime.ts b/src/agents/embedded-agent-runner/run/compaction-runtime.ts index 5087e857815..169b9807c74 100644 --- a/src/agents/embedded-agent-runner/run/compaction-runtime.ts +++ b/src/agents/embedded-agent-runner/run/compaction-runtime.ts @@ -34,6 +34,7 @@ export type EmbeddedRunCompactionRecoveryInput = { runtimeAuthPlan: Parameters[0]["runtimeAuthPlan"]; resolvedSessionKey: string; sessionAgentId: string; + contextEngineAgentId?: string; agentDir: string; workspaceDir: string; provider: string; @@ -123,7 +124,7 @@ export async function compactEmbeddedRunForRecovery( ...resolveContextEngineCapabilities({ config: runParams.config, sessionKey: runParams.sessionKey, - agentId: input.sessionAgentId, + explicitAgentId: input.contextEngineAgentId, contextEnginePluginId: input.resolveContextEnginePluginId(), purpose: recovery.trigger === "overflow" diff --git a/src/agents/embedded-agent-runner/run/execution-context.ts b/src/agents/embedded-agent-runner/run/execution-context.ts index 56bf6289691..96e1758849a 100644 --- a/src/agents/embedded-agent-runner/run/execution-context.ts +++ b/src/agents/embedded-agent-runner/run/execution-context.ts @@ -11,6 +11,7 @@ import { prepareEmbeddedRunRuntime } from "./runtime-preparation.js"; export type PreparedEmbeddedRunInput = { runParams: RunEmbeddedAgentParamsWithSessionFile; + contextEngineAgentId?: string; provider: string; modelId: string; agentDir: string; diff --git a/src/agents/embedded-agent-runner/run/overflow-context-recovery.ts b/src/agents/embedded-agent-runner/run/overflow-context-recovery.ts index 5f5657b398d..48ac0bcbcfb 100644 --- a/src/agents/embedded-agent-runner/run/overflow-context-recovery.ts +++ b/src/agents/embedded-agent-runner/run/overflow-context-recovery.ts @@ -186,6 +186,7 @@ export async function recoverEmbeddedRunOverflow( runtimeSettings: compaction.runtimeSettings, config: runParams.config, agentId: input.sessionAgentId, + contextEngineAgentId: input.contextEngineAgentId, }); } } catch (compactErr) { diff --git a/src/agents/embedded-agent-runner/run/run-attempt-dispatch.ts b/src/agents/embedded-agent-runner/run/run-attempt-dispatch.ts index 051fc301721..50fc2c689d1 100644 --- a/src/agents/embedded-agent-runner/run/run-attempt-dispatch.ts +++ b/src/agents/embedded-agent-runner/run/run-attempt-dispatch.ts @@ -34,6 +34,7 @@ type InternalRunParams = RunEmbeddedAgentParams & { }; type AttemptRuntime = { + contextEngineAgentId?: string; sessionId: string; sessionFile: string; sessionKey?: string; @@ -241,6 +242,7 @@ export async function dispatchEmbeddedRunAttempt(input: { } const attemptParams: EmbeddedRunAttemptParams = { admittedRunContext: params.admittedRunContext, + contextEngineAgentId: runtime.contextEngineAgentId, ...(control.pluginHarnessOwnsTransport ? { sandbox: pluginSandbox } : {}), operation: "attempt", sessionId: runtime.sessionId, diff --git a/src/agents/embedded-agent-runner/run/types.ts b/src/agents/embedded-agent-runner/run/types.ts index c70f5f034c7..a1a5fdc084d 100644 --- a/src/agents/embedded-agent-runner/run/types.ts +++ b/src/agents/embedded-agent-runner/run/types.ts @@ -100,6 +100,8 @@ export type EmbeddedRunAttemptTrajectoryRecorder = { export type EmbeddedRunAttemptParams = EmbeddedRunAttemptBase & { admittedRunContext: NonNullable; + /** Explicit session owner captured before fallback agent resolution. */ + contextEngineAgentId?: string; /** Host-resolved sandbox snapshot for plugin harness tool construction. */ sandbox?: SandboxContext | null; /** Host-created authority available only after harness selection. */ diff --git a/src/auto-reply/reply/commands-compact.ts b/src/auto-reply/reply/commands-compact.ts index aadb9050c78..8bb0e164fd9 100644 --- a/src/auto-reply/reply/commands-compact.ts +++ b/src/auto-reply/reply/commands-compact.ts @@ -295,6 +295,7 @@ export const handleCompactCommand: CommandHandler = async (params) => { } const result = await runtime.compactEmbeddedAgentSession({ abortSignal: params.opts?.abortSignal, + contextEngineAgentId: sessionAgentId, sessionId, sessionKey: params.sessionKey, sessionTarget: { diff --git a/src/gateway/server-methods/sessions-compaction-runner.ts b/src/gateway/server-methods/sessions-compaction-runner.ts index 5a2160a93e5..489a5cfdff9 100644 --- a/src/gateway/server-methods/sessions-compaction-runner.ts +++ b/src/gateway/server-methods/sessions-compaction-runner.ts @@ -100,6 +100,7 @@ export async function runGatewaySessionCompaction( cfg: params.cfg, }); return await compactEmbeddedAgentSession({ + contextEngineAgentId: params.agentId, sessionId: params.sessionId, sessionKey: params.sessionKey, agentId: params.agentId, diff --git a/src/plugins/runtime/runtime-llm.runtime.test.ts b/src/plugins/runtime/runtime-llm.runtime.test.ts index 47aee49a10f..f28461a7874 100644 --- a/src/plugins/runtime/runtime-llm.runtime.test.ts +++ b/src/plugins/runtime/runtime-llm.runtime.test.ts @@ -2,6 +2,8 @@ import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { resolveContextEngineCapabilities } from "../../agents/embedded-agent-runner/context-engine-capabilities.js"; +import { runContextEngineMaintenance } from "../../agents/embedded-agent-runner/context-engine-maintenance.js"; +import { buildAfterTurnRuntimeContext } from "../../agents/embedded-agent-runner/run/attempt-prompt-helpers.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { withPluginRuntimePluginIdScope } from "./gateway-request-scope.js"; import { createRuntimeLlm } from "./runtime-llm.runtime.js"; @@ -229,6 +231,69 @@ describe("runtime.llm.complete", () => { expect(hoisted.prepareSimpleCompletionModelForAgent).not.toHaveBeenCalled(); }); + it("does not trust a fallback-derived agent in the after-turn caller", async () => { + const attempt = { + sessionKey: "legacy-session", + config: cfg, + skillsSnapshot: undefined, + provider: "openai", + modelId: "gpt-5.5", + thinkLevel: "off" as const, + } satisfies Parameters[0]["attempt"]; + const runtimeContext = buildAfterTurnRuntimeContext({ + attempt, + workspaceDir: "/tmp/workspace", + agentDir: "/tmp/agent", + activeAgentId: "main", + }); + + await expect( + runtimeContext.llm!.complete({ + messages: [{ role: "user", content: "summarize" }], + }), + ).rejects.toThrow("not bound to an active session agent"); + expect(hoisted.prepareSimpleCompletionModelForAgent).not.toHaveBeenCalled(); + + const maintain = vi.fn(async (params: { runtimeContext?: typeof runtimeContext }) => { + await params.runtimeContext?.llm?.complete({ + messages: [{ role: "user", content: "maintain" }], + }); + return { changed: false, bytesFreed: 0, rewrittenEntries: 0 }; + }); + const maintenanceResult = await runContextEngineMaintenance({ + contextEngine: { + info: { id: "test", name: "Test" }, + maintain, + } as never, + sessionId: "legacy-session", + sessionKey: "legacy-session", + sessionFile: "legacy-session", + reason: "turn", + config: cfg, + agentId: "main", + runtimeContext, + }); + + expect(maintain).toHaveBeenCalledOnce(); + expect(maintenanceResult).toBeUndefined(); + expect(hoisted.prepareSimpleCompletionModelForAgent).not.toHaveBeenCalled(); + }); + + it("accepts an explicitly trusted pre-fallback agent", async () => { + const runtimeContext = resolveContextEngineCapabilities({ + config: cfg, + sessionKey: "legacy-session", + explicitAgentId: "ada", + purpose: "context-engine.after-turn", + }); + + const result = await runtimeContext.llm!.complete({ + messages: [{ role: "user", content: "summarize" }], + }); + + expect(result.agentId).toBe("ada"); + }); + it("fails closed for context-engine completions without any session agent", async () => { const runtimeContext = resolveContextEngineCapabilities({ config: cfg, diff --git a/test/scripts/docker-all-scheduler.test.ts b/test/scripts/docker-all-scheduler.test.ts index c5d865c6475..34baa2a2200 100644 --- a/test/scripts/docker-all-scheduler.test.ts +++ b/test/scripts/docker-all-scheduler.test.ts @@ -265,6 +265,14 @@ function isProcessAlive(pid: number): boolean { } } +function readCompletePidFile(pidPath: string): number | undefined { + if (!existsSync(pidPath)) { + return undefined; + } + const pid = Number.parseInt(readFileSync(pidPath, "utf8"), 10); + return Number.isInteger(pid) ? pid : undefined; +} + async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise { const started = Date.now(); while (Date.now() - started < timeoutMs) { @@ -1275,9 +1283,10 @@ setInterval(() => {}, 1000); timeoutMs: 250, }); - await waitFor(() => existsSync(grandchildPidPath)); - grandchildPid = Number.parseInt(readFileSync(grandchildPidPath, "utf8"), 10); - expect(Number.isInteger(grandchildPid)).toBe(true); + await waitFor(() => { + grandchildPid = readCompletePidFile(grandchildPidPath) ?? 0; + return grandchildPid > 0; + }); expect(isProcessAlive(grandchildPid)).toBe(true); await expect(runPromise).resolves.toMatchObject({ timedOut: true }); @@ -1494,9 +1503,10 @@ await runShellCommand({ cwd: process.cwd(), stdio: ["ignore", "ignore", "pipe"], }); - await waitFor(() => existsSync(readyPath) && existsSync(grandchildPidPath)); - grandchildPid = Number.parseInt(readFileSync(grandchildPidPath, "utf8"), 10); - expect(Number.isInteger(grandchildPid)).toBe(true); + await waitFor(() => { + grandchildPid = readCompletePidFile(grandchildPidPath) ?? 0; + return existsSync(readyPath) && grandchildPid > 0; + }); expect(isProcessAlive(grandchildPid)).toBe(true); runner.kill("SIGTERM");