opensource/kilocode
3
0
Fork 0
mirror of https://github.com/Kilo-Org/kilocode.git synced 2026-08-30 17:14:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into feat-agent-manager-orchestration-api

Browse Source
This commit is contained in:
marius-kilocode
2026-07-14 07:27:06 +02:00
parent 7337682293 e2b609eb5e
commit e604ddafcb
29 changed files with 948 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/opencode/src/kilocode/session/compaction-chunks.ts
+48 -7
View File
@@ -33,6 +33,7 @@ export namespace KiloCompactionChunks {
type Output = {
result: SessionProcessor.Result
output: string | undefined
error: MessageV2.Assistant["error"]
}
type Deps = {
@@ -272,7 +273,11 @@ export namespace KiloCompactionChunks {
model: mdl,
})
const parts = yield* MessageV2.parts(worker.message.id)
return { result, output: text(worker.message, parts) }
return {
result,
output: text(worker.message, parts),
error: worker.message.error ?? worker.compactError?.(),
}
}).pipe(
Effect.ensuring(
input.session.removeMessage({ sessionID: input.sessionID, messageID: worker.message.id }).pipe(Effect.ignore),
@@ -280,9 +285,37 @@ export namespace KiloCompactionChunks {
)
const result = out.result
const output = out.output
if (result !== "continue") return { result, output: undefined }
if (!output) return { result: "stop" as const, output: undefined }
return { result, output }
if (result !== "continue") return { result, output: undefined, error: out.error }
if (!output)
return {
result: "stop" as const,
output: undefined,
error:
out.error ??
new MessageV2.APIError({
message: "Compaction worker returned an empty response",
isRetryable: true,
}).toObject(),
}
return { result, output, error: undefined }
})
}
function fatal(output: Output | undefined) {
return output?.result === "stop" && !!output.error && output.error.name !== "ContextOverflowError"
}
function fail(input: Input, output: Output | undefined) {
return Effect.gen(function* () {
if (output?.result !== "stop") return false
const error = output.error
if (!error || error.name === "ContextOverflowError") return false
input.target.error = error
input.target.finish = "error"
input.target.time.completed = Date.now()
yield* input.updateMessage(input.target)
return true
})
}
@@ -332,7 +365,8 @@ export namespace KiloCompactionChunks {
(group) => reduce({ ...input, summaries: group, depth: input.depth + 1 }),
{ concurrency: 1 },
)
if (next.some((item) => item.result !== "continue" || !item.output)) return result
const failed = next.find(fatal) ?? next.find((item) => item.result !== "continue" || !item.output)
if (failed) return fatal(failed) ? failed : result
return yield* reduce({ ...input, summaries: next.map((item) => item.output!), depth: input.depth + 2 })
})
}
@@ -346,13 +380,20 @@ export namespace KiloCompactionChunks {
const partial = yield* Effect.forEach(chunks, (chunk) => summarize({ ...input, chunk, total: chunks.length }), {
concurrency: Math.min(CONCURRENCY, chunks.length),
})
if (partial.some((item) => item.result !== "continue" || !item.output)) return "compact" as const
const failed = partial.find(fatal) ?? partial.find((item) => item.result !== "continue" || !item.output)
if (failed) {
if (yield* fail(input, failed)) return "stop" as const
return "compact" as const
}
const final =
chunks.length === 1 && (yield* large({ messages: chunks[0].messages, model: input.model, size }))
? partial[0]
: yield* reduce({ ...input, summaries: partial.map((item) => item.output!), depth: 0 })
if (!final || final.result !== "continue" || !final.output) return "compact" as const
if (!final || final.result !== "continue" || !final.output) {
if (yield* fail(input, final)) return "stop" as const
return "compact" as const
}
yield* input.updatePart({
id: PartID.ascending(),
packages/opencode/src/session/prompt.ts
+7 -1
View File
@@ -769,6 +769,8 @@ export const layer = Layer.effect(
.get()
.pipe(Effect.orDie)
const model = input.model ?? ag.model ?? (yield* currentModel(input.sessionID))
// kilocode_change start - retain the source session variant across Agent Manager's model-less fork handoff
const stored = !input.model && !ag.model ? model : undefined
const same = ag.model && model.providerID === ag.model.providerID && model.modelID === ag.model.modelID
const full =
!input.variant && ag.variant && same
@@ -776,7 +778,11 @@ export const layer = Layer.effect(
.getModel(model.providerID, model.modelID)
.pipe(Effect.catchIf(Provider.ModelNotFoundError.isInstance, () => Effect.succeed(undefined)))
: undefined
const variant = input.variant ?? (ag.variant && full?.variants?.[ag.variant] ? ag.variant : undefined)
const variant =
input.variant ??
(stored && "variant" in stored && typeof stored.variant === "string" ? stored.variant : undefined) ??
(ag.variant && full?.variants?.[ag.variant] ? ag.variant : undefined)
// kilocode_change end
const info: SessionV1.User = {
id: input.messageID ?? MessageID.ascending(),
packages/opencode/src/session/session.ts
+20 -1
View File
@@ -803,16 +803,35 @@ export const layer: Layer.Layer<
// kilocode_change start - forks into another directory cannot read the source confinement from the new dir, so carry it over explicitly
const sandboxFallback = yield* SandboxPolicy.peek(original.directory, input.sessionID)
// kilocode_change end
// kilocode_change start - historical forks must use the model from retained context, not a later source-session selection
const msgs = yield* messages({ sessionID: input.sessionID })
const point = input.messageID
const message = point
? msgs.findLast((msg) => msg.info.id < point && msg.info.role === "user")
: undefined
const model =
message?.info.role === "user"
? {
id: message.info.model.modelID,
providerID: message.info.model.providerID,
variant: message.info.model.variant,
}
: point
? undefined
: original.model
? { ...original.model }
: undefined
// kilocode_change end
const session = yield* createNext({
directory: ctx.directory,
path: sessionPath(ctx.worktree, ctx.directory),
workspaceID: original.workspaceID,
title,
metadata: structuredClone(original.metadata),
model, // kilocode_change - preserve the model + variant active at the fork point
sourceID: input.sessionID, // kilocode_change - forks preserve initialized confinement
sandboxFallback, // kilocode_change - seed confinement from the source session's original directory
})
const msgs = yield* messages({ sessionID: input.sessionID })
const idMap = new Map<string, MessageID>()
for (const msg of msgs) {
packages/opencode/test/kilocode/session-compaction-chunks.test.ts
+104 -8
View File
@@ -168,7 +168,7 @@ function reply(text: string, capture?: (input: LLM.StreamInput) => void) {
}
}
function fakeRuntime(outputTokenMax?: number) {
function fakeRuntime(outputTokenMax?: number, error?: MessageV2.Assistant["error"], empty = false) {
const calls: string[] = []
const outputs: number[] = []
const bus = Bus.layer
@@ -189,6 +189,12 @@ function fakeRuntime(outputTokenMax?: number) {
Effect.gen(function* () {
outputs.push(input.model.limit.output)
calls.push(JSON.stringify(stream.messages))
if (error) {
input.assistantMessage.error = error
input.assistantMessage.finish = "error"
yield* sessions.updateMessage(input.assistantMessage)
return "stop" as const
}
const text = stream.messages.some((msg) =>
JSON.stringify(msg).includes("Create a new anchored summary"),
)
@@ -196,13 +202,14 @@ function fakeRuntime(outputTokenMax?: number) {
: calls.length === 1
? "chunk one"
: "chunk two"
yield* sessions.updatePart({
id: PartID.ascending(),
messageID: input.assistantMessage.id,
sessionID: input.sessionID,
type: "text",
text,
})
if (!empty)
yield* sessions.updatePart({
id: PartID.ascending(),
messageID: input.assistantMessage.id,
sessionID: input.sessionID,
type: "text",
text,
})
input.assistantMessage.finish = "stop"
return "continue" as const
}),
@@ -238,6 +245,48 @@ function fakeRuntime(outputTokenMax?: number) {
}
}
async function failure(error?: MessageV2.Assistant["error"], empty = false) {
await using tmp = await tmpdir()
return provideTestInstance({
directory: tmp.path,
fn: async () => {
const session = await svc.create({})
await user(session.id, "oversized " + "x".repeat(80_000))
await Effect.runPromise(
KiloSessionCompaction.create({
session: store,
sessionID: session.id,
agent: "build",
model: ref,
auto: false,
}),
)
const { rt } = fakeRuntime(undefined, error, empty)
try {
const msgs = await svc.messages({ sessionID: session.id })
const parent = msgs.at(-1)?.info.id
expect(parent).toBeTruthy()
const result = await rt.runPromise(
SessionCompaction.Service.use((svc) =>
svc.process({
parentID: parent!,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
)
const all = await svc.messages({ sessionID: session.id })
const summary = all.find((msg) => msg.info.role === "assistant" && msg.info.summary)
return { result, summary }
} finally {
await rt.dispose()
}
},
})
}
function liveRuntime(layer: Layer.Layer<LLM.Service>, context = 10_000) {
const bus = Bus.layer
const status = SessionStatus.layer.pipe(Layer.provide(bus), Layer.provide(EventV2Bridge.defaultLayer))
@@ -317,6 +366,53 @@ describe("KiloCompactionChunks", () => {
expect(KiloCompactionChunks.budget({ cfg, model, outputTokenMax })).toBe(5_692)
})
test("preserves gateway errors from chunk workers", async () => {
const error = new MessageV2.APIError({
message: "The operation was aborted",
statusCode: 504,
isRetryable: true,
responseBody: '{"error_type":"timeout"}',
}).toObject()
const result = await failure(error)
expect(result.result).toBe("stop")
expect(result.summary?.info.role).toBe("assistant")
if (result.summary?.info.role !== "assistant") return
expect(result.summary.info.finish).toBe("error")
expect(result.summary.info.error).toEqual(error)
})
test("keeps context overflow on the terminal compaction path", async () => {
const result = await failure(
new MessageV2.ContextOverflowError({
message: "worker context overflow",
}).toObject(),
)
expect(result.result).toBe("stop")
expect(result.summary?.info.role).toBe("assistant")
if (result.summary?.info.role !== "assistant") return
expect(result.summary.info.error?.name).toBe("ContextOverflowError")
if (result.summary.info.error?.name !== "ContextOverflowError") return
expect(result.summary.info.error.data.message).toBe(
"Session too large to compact - context exceeds model limit even after stripping media",
)
})
test("reports empty chunk worker responses as API errors", async () => {
const result = await failure(undefined, true)
expect(result.result).toBe("stop")
expect(result.summary?.info.role).toBe("assistant")
if (result.summary?.info.role !== "assistant") return
expect(result.summary.info.finish).toBe("error")
expect(result.summary.info.error?.name).toBe("APIError")
if (result.summary.info.error?.name !== "APIError") return
expect(result.summary.info.error.data.message).toBe("Compaction worker returned an empty response")
expect(result.summary.info.error.data.isRetryable).toBe(true)
})
test("falls back to chunk workers after the first compaction overflows", async () => {
await using tmp = await tmpdir()
await provideTestInstance({
packages/opencode/test/kilocode/session-overflow.test.ts
+13
View File
@@ -123,6 +123,19 @@ describe("Kilo auto-compaction threshold", () => {
expect(isOverflow({ cfg: conf, model: mdl, tokens: { ...tokens(0), total: 150_000 } })).toBe(true)
})
test("uses the output cap as the reserve for single-window gateway models", () => {
const mdl = model({ context: 262_144, output: 262_144 })
expect(usable({ cfg: cfg(), model: mdl })).toBe(230_144)
expect(usable({ cfg: cfg({ reserved: 20_000 }), model: mdl })).toBe(230_144)
})
test("keeps usable context for small single-window models with large output limits", () => {
const mdl = model({ context: 40_000, output: 262_144 })
expect(usable({ cfg: cfg(), model: mdl })).toBe(8_000)
})
})
describe("Kilo request estimation", () => {
packages/opencode/test/session/prompt.test.ts
+35
View File
@@ -2532,6 +2532,41 @@ it.instance(
// Agent variant
// kilocode_change start - Agent Manager records a model-less synthetic prompt after forking
noLLMServer.instance(
"preserves the session variant through a model-less handoff",
() =>
Effect.gen(function* () {
const prompt = yield* SessionPrompt.Service
const sessions = yield* Session.Service
const session = yield* sessions.create({
model: {
id: ref.modelID,
providerID: ref.providerID,
variant: "high",
},
})
const handoff = yield* prompt.prompt({
sessionID: session.id,
noReply: true,
parts: [{ type: "text", text: "fork handoff", synthetic: true }],
})
if (handoff.info.role !== "user") throw new Error("expected user message")
expect(handoff.info.model).toEqual({
providerID: ref.providerID,
modelID: ref.modelID,
variant: "high",
})
const saved = yield* sessions.get(session.id)
expect(saved.model?.variant).toBe("high")
}),
{ config: cfg },
)
// kilocode_change end
noLLMServer.instance(
"applies agent variant only when using agent model",
() =>
packages/opencode/test/session/session.test.ts
+85
View File
@@ -8,6 +8,7 @@ import { Session as SessionNs } from "@/session/session"
import * as Log from "@opencode-ai/core/util/log"
import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, type SessionID } from "../../src/session/schema"
type SessionModel = NonNullable<SessionNs.Info["model"]> // kilocode_change
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { provideInstance, testInstanceStoreLayer, tmpdirScoped } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
@@ -248,4 +249,88 @@ describe("Session", () => {
expect(saved.metadata).toBeUndefined()
}),
)
// kilocode_change start
it.instance("fork preserves model and variant", () =>
Effect.gen(function* () {
const session = yield* SessionNs.Service
const model = {
id: "test-model",
providerID: "test-provider",
variant: "high",
} as SessionModel
const created = yield* Effect.acquireRelease(
session.create({ title: "with-model", model }),
(info) => session.remove(info.id).pipe(Effect.ignore),
)
const saved = yield* session.get(created.id)
expect(saved.model).toEqual(model)
const fork = yield* Effect.acquireRelease(session.fork({ sessionID: created.id }), (info) =>
session.remove(info.id).pipe(Effect.ignore),
)
const forked = yield* session.get(fork.id)
expect(forked.model).toEqual(model)
expect(forked.model?.variant).toBe("high")
expect(forked.model).not.toBe(saved.model)
}),
)
// kilocode_change end
// kilocode_change start
it.instance("historical fork preserves the model at the fork point", () =>
Effect.gen(function* () {
const session = yield* SessionNs.Service
const source = yield* Effect.acquireRelease(
session.create({
model: {
id: "test-model",
providerID: "test-provider",
variant: "high",
} as SessionModel,
}),
(info) => session.remove(info.id).pipe(Effect.ignore),
)
yield* session.updateMessage({
id: MessageID.ascending(),
sessionID: source.id,
role: "user",
time: { created: Date.now() },
agent: "code",
model: {
providerID: source.model!.providerID,
modelID: source.model!.id,
variant: "low",
},
tools: {},
mode: "",
} as unknown as MessageV2.Info)
const latest = yield* session.updateMessage({
id: MessageID.ascending(),
sessionID: source.id,
role: "user",
time: { created: Date.now() },
agent: "code",
model: {
providerID: source.model!.providerID,
modelID: source.model!.id,
variant: "high",
},
tools: {},
mode: "",
} as unknown as MessageV2.Info)
const fork = yield* Effect.acquireRelease(
session.fork({ sessionID: source.id, messageID: latest.id }),
(info) => session.remove(info.id).pipe(Effect.ignore),
)
expect(fork.model).toEqual({
id: source.model!.id,
providerID: source.model!.providerID,
variant: "low",
})
}),
)
// kilocode_change end
})