ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): narrow v2 context epoch scope

Browse source
This commit is contained in:
Kit Langton 2026-06-04 21:25:46 -04:00
commit d856d92506
11 changed files with 339 additions and 827 deletions
Download patch file Download diff file Expand all files Collapse all files

1
AGENTS.md
View file

@ -133,7 +133,6 @@ const table = sqliteTable("session", {
- Avoid mocks as much as possible - Avoid mocks as much as possible
- Test actual implementation, do not duplicate logic into tests - Test actual implementation, do not duplicate logic into tests
- In `packages/core/test`, define tests with the shared `it.effect` or `testEffect(...)` helpers from `test/lib/effect.ts`; do not use raw `test(...)`. Wrap synchronous assertions in `Effect.sync(...)`.
- Tests cannot run from repo root (guard: `do-not-run-tests-from-root`); run from package dirs like `packages/opencode`. - Tests cannot run from repo root (guard: `do-not-run-tests-from-root`); run from package dirs like `packages/opencode`.
## Type Checking ## Type Checking

670
packages/core/test/session-runner-message.test.ts
View file

@ -18,74 +18,74 @@ const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.
describe("toLLMMessages", () => { describe("toLLMMessages", () => {
it.effect("maps every top-level V2 Session message type", () => it.effect("maps every top-level V2 Session message type", () =>
Effect.sync(() => { Effect.sync(() => {
const file = new FileAttachment({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" }) const file = new FileAttachment({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" })
const reference = new ReferenceAttachment({ name: "docs", kind: "local", uri: "file:///docs" }) const reference = new ReferenceAttachment({ name: "docs", kind: "local", uri: "file:///docs" })
const messages = toLLMMessages( const messages = toLLMMessages(
[ [
new SessionMessage.AgentSwitched({ new SessionMessage.AgentSwitched({
id: id("agent"), id: id("agent"),
type: "agent-switched", type: "agent-switched",
agent: "build", agent: "build",
time: { created }, time: { created },
}),
new SessionMessage.ModelSwitched({
id: id("model"),
type: "model-switched",
model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") },
time: { created },
}),
new SessionMessage.User({
id: id("user"),
type: "user",
text: "Inspect this image",
files: [file],
agents: [new AgentAttachment({ name: "build" })],
references: [reference],
time: { created },
}),
new SessionMessage.Synthetic({
id: id("synthetic"),
type: "synthetic",
sessionID: SessionV2.ID.make("ses_translate"),
text: "Synthetic context",
time: { created },
}),
new SessionMessage.Shell({
id: id("shell"),
type: "shell",
callID: "shell-1",
command: "pwd",
output: "/project",
time: { created, completed: created },
}),
new SessionMessage.Compaction({
id: id("compaction"),
type: "compaction",
reason: "auto",
summary: "Earlier work",
time: { created },
}),
],
model,
)
expect(messages.map((message) => message.role)).toEqual(["user", "user", "user", "user"])
expect(messages[0]).toEqual(
Message.make({
id: id("user"),
role: "user",
content: [
{ type: "text", text: "Inspect this image" },
{ type: "media", mediaType: "image/png", data: "data:image/png;base64,aGVsbG8=", filename: "hello.png" },
],
metadata: { agents: [{ name: "build" }], references: [reference] },
}), }),
) new SessionMessage.ModelSwitched({
expect(messages.slice(1).map((message) => message.content)).toEqual([ id: id("model"),
[{ type: "text", text: "Synthetic context" }], type: "model-switched",
[{ type: "text", text: "Shell command: pwd\n\n/project" }], model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") },
[{ type: "text", text: "Summary of earlier conversation:\nEarlier work" }], time: { created },
]) }),
new SessionMessage.User({
id: id("user"),
type: "user",
text: "Inspect this image",
files: [file],
agents: [new AgentAttachment({ name: "build" })],
references: [reference],
time: { created },
}),
new SessionMessage.Synthetic({
id: id("synthetic"),
type: "synthetic",
sessionID: SessionV2.ID.make("ses_translate"),
text: "Synthetic context",
time: { created },
}),
new SessionMessage.Shell({
id: id("shell"),
type: "shell",
callID: "shell-1",
command: "pwd",
output: "/project",
time: { created, completed: created },
}),
new SessionMessage.Compaction({
id: id("compaction"),
type: "compaction",
reason: "auto",
summary: "Earlier work",
time: { created },
}),
],
model,
)
expect(messages.map((message) => message.role)).toEqual(["user", "user", "user", "user"])
expect(messages[0]).toEqual(
Message.make({
id: id("user"),
role: "user",
content: [
{ type: "text", text: "Inspect this image" },
{ type: "media", mediaType: "image/png", data: "data:image/png;base64,aGVsbG8=", filename: "hello.png" },
],
metadata: { agents: [{ name: "build" }], references: [reference] },
}),
)
expect(messages.slice(1).map((message) => message.content)).toEqual([
[{ type: "text", text: "Synthetic context" }],
[{ type: "text", text: "Shell command: pwd\n\n/project" }],
[{ type: "text", text: "Summary of earlier conversation:\nEarlier work" }],
])
}), }),
) )
@ -109,295 +109,295 @@ describe("toLLMMessages", () => {
it.effect("expands assistant tool calls and settled outcomes into canonical tool messages", () => it.effect("expands assistant tool calls and settled outcomes into canonical tool messages", () =>
Effect.sync(() => { Effect.sync(() => {
const messages = toLLMMessages( const messages = toLLMMessages(
[ [
new SessionMessage.Assistant({ new SessionMessage.Assistant({
id: id("assistant"), id: id("assistant"),
type: "assistant", type: "assistant",
agent: "build", agent: "build",
model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") },
content: [ content: [
new SessionMessage.AssistantText({ type: "text", id: "text-1", text: "Checking" }), new SessionMessage.AssistantText({ type: "text", id: "text-1", text: "Checking" }),
new SessionMessage.AssistantReasoning({ new SessionMessage.AssistantReasoning({
type: "reasoning", type: "reasoning",
id: "reasoning-1", id: "reasoning-1",
text: "Think", text: "Think",
providerMetadata: { anthropic: { signature: "sig_1" } }, providerMetadata: { anthropic: { signature: "sig_1" } },
}),
new SessionMessage.AssistantTool({
type: "tool",
id: "pending",
name: "read",
state: new SessionMessage.ToolStatePending({ status: "pending", input: '{"path":"README.md"}' }),
time: { created },
}),
new SessionMessage.AssistantTool({
type: "tool",
id: "running",
name: "read",
state: new SessionMessage.ToolStateRunning({
status: "running",
input: { path: "README.md" },
content: [],
structured: {},
}), }),
new SessionMessage.AssistantTool({ time: { created },
type: "tool", }),
id: "pending", new SessionMessage.AssistantTool({
name: "read", type: "tool",
state: new SessionMessage.ToolStatePending({ status: "pending", input: '{"path":"README.md"}' }), id: "completed",
time: { created }, name: "read",
state: new SessionMessage.ToolStateCompleted({
status: "completed",
input: { path: "README.md" },
content: [
new ToolOutput.TextContent({ type: "text", text: "Hello" }),
new ToolOutput.FileContent({
type: "file",
source: { type: "data", data: "aGVsbG8=" },
mime: "image/png",
name: "hello.png",
}),
],
structured: {},
}), }),
new SessionMessage.AssistantTool({ time: { created, completed: created },
type: "tool", }),
id: "running", new SessionMessage.AssistantTool({
name: "read", type: "tool",
state: new SessionMessage.ToolStateRunning({ id: "hosted",
status: "running", name: "web_search",
input: { path: "README.md" }, provider: {
content: [], executed: true,
structured: {}, metadata: { fake: { continuation: "hosted-call" } },
}), resultMetadata: { fake: { continuation: "hosted-result" } },
time: { created }, },
state: new SessionMessage.ToolStateCompleted({
status: "completed",
input: { query: "Effect" },
content: [new ToolOutput.TextContent({ type: "text", text: "Found it" })],
structured: {},
}), }),
new SessionMessage.AssistantTool({ time: { created, completed: created },
type: "tool", }),
id: "completed", new SessionMessage.AssistantTool({
name: "read", type: "tool",
state: new SessionMessage.ToolStateCompleted({ id: "hosted-failed",
status: "completed", name: "write",
input: { path: "README.md" }, provider: { executed: true, metadata: { fake: { continuation: "failed" } } },
content: [ state: new SessionMessage.ToolStateError({
new ToolOutput.TextContent({ type: "text", text: "Hello" }), status: "error",
new ToolOutput.FileContent({ input: { path: "README.md" },
type: "file", content: [],
source: { type: "data", data: "aGVsbG8=" }, structured: {},
mime: "image/png", error: { type: "unknown", message: "Denied" },
name: "hello.png",
}),
],
structured: {},
}),
time: { created, completed: created },
}), }),
new SessionMessage.AssistantTool({ time: { created, completed: created },
type: "tool", }),
id: "hosted", ],
name: "web_search", time: { created, completed: created },
provider: { }),
executed: true, ],
metadata: { fake: { continuation: "hosted-call" } }, model,
resultMetadata: { fake: { continuation: "hosted-result" } }, )
},
state: new SessionMessage.ToolStateCompleted({
status: "completed",
input: { query: "Effect" },
content: [new ToolOutput.TextContent({ type: "text", text: "Found it" })],
structured: {},
}),
time: { created, completed: created },
}),
new SessionMessage.AssistantTool({
type: "tool",
id: "hosted-failed",
name: "write",
provider: { executed: true, metadata: { fake: { continuation: "failed" } } },
state: new SessionMessage.ToolStateError({
status: "error",
input: { path: "README.md" },
content: [],
structured: {},
error: { type: "unknown", message: "Denied" },
}),
time: { created, completed: created },
}),
],
time: { created, completed: created },
}),
],
model,
)
expect(messages.map((message) => message.role)).toEqual(["assistant", "tool"]) expect(messages.map((message) => message.role)).toEqual(["assistant", "tool"])
expect(messages[0]?.content).toEqual([ expect(messages[0]?.content).toEqual([
{ type: "text", text: "Checking" }, { type: "text", text: "Checking" },
{ type: "reasoning", text: "Think", providerMetadata: { anthropic: { signature: "sig_1" } } }, { type: "reasoning", text: "Think", providerMetadata: { anthropic: { signature: "sig_1" } } },
{ type: "tool-call", id: "pending", name: "read", input: { path: "README.md" } }, { type: "tool-call", id: "pending", name: "read", input: { path: "README.md" } },
{ type: "tool-call", id: "running", name: "read", input: { path: "README.md" } }, { type: "tool-call", id: "running", name: "read", input: { path: "README.md" } },
{ {
type: "tool-call", type: "tool-call",
id: "completed", id: "completed",
name: "read", name: "read",
input: { path: "README.md" }, input: { path: "README.md" },
},
{
type: "tool-call",
id: "hosted",
name: "web_search",
input: { query: "Effect" },
providerExecuted: true,
providerMetadata: { fake: { continuation: "hosted-call" } },
},
{
type: "tool-result",
id: "hosted",
name: "web_search",
providerExecuted: true,
providerMetadata: { fake: { continuation: "hosted-result" } },
result: { type: "text", value: "Found it" },
},
{
type: "tool-call",
id: "hosted-failed",
name: "write",
input: { path: "README.md" },
providerExecuted: true,
providerMetadata: { fake: { continuation: "failed" } },
},
{
type: "tool-result",
id: "hosted-failed",
name: "write",
providerExecuted: true,
providerMetadata: { fake: { continuation: "failed" } },
result: {
type: "error",
value: { error: { type: "unknown", message: "Denied" }, content: [], structured: {} },
}, },
{ },
type: "tool-call", ])
id: "hosted", expect(messages[1]?.content).toEqual([
name: "web_search", {
input: { query: "Effect" }, type: "tool-result",
providerExecuted: true, id: "completed",
providerMetadata: { fake: { continuation: "hosted-call" } }, name: "read",
result: {
type: "content",
value: [
{ type: "text", text: "Hello" },
{ type: "media", mediaType: "image/png", data: "aGVsbG8=", filename: "hello.png" },
],
}, },
{ },
type: "tool-result", ])
id: "hosted",
name: "web_search",
providerExecuted: true,
providerMetadata: { fake: { continuation: "hosted-result" } },
result: { type: "text", value: "Found it" },
},
{
type: "tool-call",
id: "hosted-failed",
name: "write",
input: { path: "README.md" },
providerExecuted: true,
providerMetadata: { fake: { continuation: "failed" } },
},
{
type: "tool-result",
id: "hosted-failed",
name: "write",
providerExecuted: true,
providerMetadata: { fake: { continuation: "failed" } },
result: {
type: "error",
value: { error: { type: "unknown", message: "Denied" }, content: [], structured: {} },
},
},
])
expect(messages[1]?.content).toEqual([
{
type: "tool-result",
id: "completed",
name: "read",
result: {
type: "content",
value: [
{ type: "text", text: "Hello" },
{ type: "media", mediaType: "image/png", data: "aGVsbG8=", filename: "hello.png" },
],
},
},
])
}), }),
) )
it.effect("restores OpenAI encrypted reasoning metadata", () => it.effect("restores OpenAI encrypted reasoning metadata", () =>
Effect.sync(() => { Effect.sync(() => {
const messages = toLLMMessages( const messages = toLLMMessages(
[ [
new SessionMessage.Assistant({ new SessionMessage.Assistant({
id: id("assistant-openai-reasoning"), id: id("assistant-openai-reasoning"),
type: "assistant", type: "assistant",
agent: "build", agent: "build",
model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") },
content: [ content: [
new SessionMessage.AssistantReasoning({ new SessionMessage.AssistantReasoning({
type: "reasoning", type: "reasoning",
id: "reasoning-openai", id: "reasoning-openai",
text: "Think", text: "Think",
providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } },
}), }),
], ],
time: { created, completed: created }, time: { created, completed: created },
}), }),
], ],
model, model,
) )
expect(messages[0]?.content).toEqual([ expect(messages[0]?.content).toEqual([
{ {
type: "reasoning", type: "reasoning",
text: "Think", text: "Think",
providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } },
}, },
]) ])
}), }),
) )
it.effect("drops provider-native continuation metadata after a model switch", () => it.effect("drops provider-native continuation metadata after a model switch", () =>
Effect.sync(() => { Effect.sync(() => {
const messages = toLLMMessages( const messages = toLLMMessages(
[ [
new SessionMessage.Assistant({ new SessionMessage.Assistant({
id: id("assistant-old-model"), id: id("assistant-old-model"),
type: "assistant", type: "assistant",
agent: "build", agent: "build",
model: { id: ModelV2.ID.make("old-model"), providerID: ProviderV2.ID.make("provider") }, model: { id: ModelV2.ID.make("old-model"), providerID: ProviderV2.ID.make("provider") },
content: [ content: [
new SessionMessage.AssistantReasoning({ new SessionMessage.AssistantReasoning({
type: "reasoning", type: "reasoning",
id: "reasoning-old-model", id: "reasoning-old-model",
text: "Visible thought", text: "Visible thought",
providerMetadata: { anthropic: { signature: "sig_old" } }, providerMetadata: { anthropic: { signature: "sig_old" } },
}),
new SessionMessage.AssistantTool({
type: "tool",
id: "hosted-old-model",
name: "web_search",
provider: {
executed: true,
metadata: { openai: { itemId: "hosted-old-model" } },
resultMetadata: { openai: { itemId: "hosted-old-model" } },
},
state: new SessionMessage.ToolStateCompleted({
status: "completed",
input: { query: "Effect" },
content: [],
structured: {},
result: { type: "json", value: { status: "completed" } },
}), }),
new SessionMessage.AssistantTool({ time: { created, completed: created },
type: "tool", }),
id: "hosted-old-model", new SessionMessage.AssistantTool({
name: "web_search", type: "tool",
provider: { id: "local-old-model",
executed: true, name: "read",
metadata: { openai: { itemId: "hosted-old-model" } }, provider: {
resultMetadata: { openai: { itemId: "hosted-old-model" } }, executed: false,
}, metadata: { fake: { call: "old" } },
state: new SessionMessage.ToolStateCompleted({ resultMetadata: { fake: { result: "old" } },
status: "completed", },
input: { query: "Effect" }, state: new SessionMessage.ToolStateCompleted({
content: [], status: "completed",
structured: {}, input: { path: "README.md" },
result: { type: "json", value: { status: "completed" } }, content: [],
}), structured: { text: "Hello" },
time: { created, completed: created },
}), }),
new SessionMessage.AssistantTool({ time: { created, completed: created },
type: "tool", }),
id: "local-old-model", ],
name: "read", time: { created, completed: created },
provider: { }),
executed: false, ],
metadata: { fake: { call: "old" } }, model,
resultMetadata: { fake: { result: "old" } }, )
},
state: new SessionMessage.ToolStateCompleted({
status: "completed",
input: { path: "README.md" },
content: [],
structured: { text: "Hello" },
}),
time: { created, completed: created },
}),
],
time: { created, completed: created },
}),
],
model,
)
expect(messages[0]?.content).toEqual([ expect(messages[0]?.content).toEqual([
{ type: "text", text: "Visible thought" }, { type: "text", text: "Visible thought" },
{ {
type: "tool-call", type: "tool-call",
id: "hosted-old-model", id: "hosted-old-model",
name: "web_search", name: "web_search",
input: { query: "Effect" }, input: { query: "Effect" },
providerExecuted: true, providerExecuted: true,
providerMetadata: undefined, providerMetadata: undefined,
}, },
{ {
type: "tool-result", type: "tool-result",
id: "hosted-old-model", id: "hosted-old-model",
name: "web_search", name: "web_search",
result: { type: "json", value: { status: "completed" } }, result: { type: "json", value: { status: "completed" } },
providerExecuted: true, providerExecuted: true,
cache: undefined, cache: undefined,
metadata: undefined, metadata: undefined,
providerMetadata: undefined, providerMetadata: undefined,
}, },
{ {
type: "tool-call", type: "tool-call",
id: "local-old-model", id: "local-old-model",
name: "read", name: "read",
input: { path: "README.md" }, input: { path: "README.md" },
providerExecuted: false, providerExecuted: false,
providerMetadata: undefined, providerMetadata: undefined,
}, },
]) ])
expect(messages[1]?.content).toEqual([ expect(messages[1]?.content).toEqual([
{ {
type: "tool-result", type: "tool-result",
id: "local-old-model", id: "local-old-model",
name: "read", name: "read",
result: { type: "json", value: { text: "Hello" } }, result: { type: "json", value: { text: "Hello" } },
providerExecuted: false, providerExecuted: false,
cache: undefined, cache: undefined,
metadata: undefined, metadata: undefined,
providerMetadata: undefined, providerMetadata: undefined,
}, },
]) ])
}), }),
) )
}) })

1
packages/llm/src/protocols/anthropic-messages.ts
View file

@ -386,7 +386,6 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (
for (const [index, message] of request.messages.entries()) { for (const [index, message] of request.messages.entries()) {
if (message.role === "system") { if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("Anthropic Messages", request.messages, index)
if (supportsNativeSystemUpdates(request) && canUseNativeSystemUpdate(request.messages, index)) { if (supportsNativeSystemUpdates(request) && canUseNativeSystemUpdate(request.messages, index)) {
messages.push(yield* lowerNativeSystemUpdate(message, breakpoints)) messages.push(yield* lowerNativeSystemUpdate(message, breakpoints))
continue continue

3
packages/llm/src/protocols/bedrock-converse.ts
View file

@ -292,9 +292,8 @@ const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (
) { ) {
const messages: BedrockMessage[] = [] const messages: BedrockMessage[] = []
for (const [index, message] of request.messages.entries()) { for (const message of request.messages) {
if (message.role === "system") { if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("Bedrock Converse", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("Bedrock Converse", message) const part = yield* ProviderShared.wrappedSystemUpdate("Bedrock Converse", message)
const content = textWithCache(breakpoints, part.text, part.cache) const content = textWithCache(breakpoints, part.text, part.cache)
const previous = messages.at(-1) const previous = messages.at(-1)

3
packages/llm/src/protocols/gemini.ts
View file

@ -200,9 +200,8 @@ const lowerToolCall = (part: ToolCallPart) => ({
const lowerMessages = Effect.fn("Gemini.lowerMessages")(function* (request: LLMRequest) { const lowerMessages = Effect.fn("Gemini.lowerMessages")(function* (request: LLMRequest) {
const contents: GeminiContent[] = [] const contents: GeminiContent[] = []
for (const [index, message] of request.messages.entries()) { for (const message of request.messages) {
if (message.role === "system") { if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("Gemini", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("Gemini", message) const part = yield* ProviderShared.wrappedSystemUpdate("Gemini", message)
const previous = contents.at(-1) const previous = contents.at(-1)
if (previous?.role === "user") if (previous?.role === "user")

3
packages/llm/src/protocols/openai-chat.ts
View file

@ -252,9 +252,8 @@ const lowerMessages = Effect.fn("OpenAIChat.lowerMessages")(function* (request:
const system: OpenAIChatMessage[] = const system: OpenAIChatMessage[] =
request.system.length === 0 ? [] : [{ role: "system", content: ProviderShared.joinText(request.system) }] request.system.length === 0 ? [] : [{ role: "system", content: ProviderShared.joinText(request.system) }]
const messages = [...system] const messages = [...system]
for (const [index, message] of request.messages.entries()) { for (const message of request.messages) {
if (message.role === "system") { if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("OpenAI Chat", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("OpenAI Chat", message) const part = yield* ProviderShared.wrappedSystemUpdate("OpenAI Chat", message)
const previous = messages.at(-1) const previous = messages.at(-1)
if (previous?.role === "user") if (previous?.role === "user")

3
packages/llm/src/protocols/openai-responses.ts
View file

@ -338,9 +338,8 @@ const lowerMessages = Effect.fn("OpenAIResponses.lowerMessages")(function* (requ
const input: OpenAIResponsesInputItem[] = [...system] const input: OpenAIResponsesInputItem[] = [...system]
const store = OpenAIOptions.store(request) const store = OpenAIOptions.store(request)
for (const [index, message] of request.messages.entries()) { for (const message of request.messages) {
if (message.role === "system") { if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("OpenAI Responses", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("OpenAI Responses", message) const part = yield* ProviderShared.wrappedSystemUpdate("OpenAI Responses", message)
const previous = input.at(-1) const previous = input.at(-1)
if (previous && "role" in previous && previous.role === "user") if (previous && "role" in previous && previous.role === "user")

20
packages/llm/src/protocols/shared.ts
View file

@ -177,26 +177,6 @@ export const wrappedSystemUpdate = Effect.fn("ProviderShared.wrappedSystemUpdate
return { type: "text" as const, text: wrapSystemUpdate(content), cache: content.at(-1)?.cache } return { type: "text" as const, text: wrapSystemUpdate(content), cache: content.at(-1)?.cache }
}) })
export const guardSystemUpdatePlacement = Effect.fn("ProviderShared.guardSystemUpdatePlacement")(function* (
route: string,
messages: LLMRequest["messages"],
index: number,
) {
const pending = new Set<string>()
for (const message of messages.slice(0, index)) {
for (const part of message.content) {
if (message.role === "assistant" && part.type === "tool-call" && part.providerExecuted !== true)
pending.add(part.id)
if (message.role === "tool" && part.type === "tool-result") pending.delete(part.id)
}
}
if (pending.size > 0)
return yield* invalidRequest(
`${route} chronological system updates cannot appear between a local tool call and its tool result`,
)
return yield* Effect.void
})
/** /**
* Parse the streamed JSON input of a tool call. Treats an empty string as * Parse the streamed JSON input of a tool call. Treats an empty string as
* `"{}"` — providers occasionally finish a tool call without ever emitting * `"{}"` — providers occasionally finish a tool call without ever emitting

14
packages/llm/test/provider/anthropic-messages.test.ts
View file

@ -127,9 +127,6 @@ describe("Anthropic Messages route", () => {
it.effect("falls back for unsupported native chronological system update placement", () => it.effect("falls back for unsupported native chronological system update placement", () =>
Effect.gen(function* () { Effect.gen(function* () {
const placementError = (messages: Parameters<typeof LLM.request>[0]["messages"]) =>
LLMClient.prepare(LLM.request({ model: opus48, messages, cache: "none" })).pipe(Effect.flip)
expect( expect(
(yield* LLMClient.prepare<AnthropicMessages.AnthropicMessagesBody>( (yield* LLMClient.prepare<AnthropicMessages.AnthropicMessagesBody>(
LLM.request({ LLM.request({
@ -168,17 +165,6 @@ describe("Anthropic Messages route", () => {
], ],
}, },
]) ])
expect(
(yield* placementError([
Message.user("Use the tool."),
Message.assistant([
ToolCallPart.make({ id: "call_1", name: "lookup", input: {} }),
{ type: "text", text: "Waiting." },
]),
Message.system("Too early."),
Message.tool({ id: "call_1", name: "lookup", result: "Done." }),
])).message,
).toContain("cannot appear between a local tool call and its tool result")
}), }),
) )

30
packages/llm/test/schema.test.ts
View file

@ -1,5 +1,5 @@
import { describe, expect, test } from "bun:test" import { describe, expect, test } from "bun:test"
import { Effect, Schema } from "effect" import { Schema } from "effect"
import * as OpenAIChat from "../src/protocols/openai-chat" import * as OpenAIChat from "../src/protocols/openai-chat"
import * as OpenAIResponses from "../src/protocols/openai-responses" import * as OpenAIResponses from "../src/protocols/openai-responses"
import { import {
@ -10,7 +10,6 @@ import {
Model, Model,
ModelID, ModelID,
ProviderID, ProviderID,
ToolCallPart,
Usage, Usage,
} from "../src/schema" } from "../src/schema"
import { ProviderShared } from "../src/protocols/shared" import { ProviderShared } from "../src/protocols/shared"
@ -64,33 +63,6 @@ describe("llm schema", () => {
expect(decoded.messages[0]).toMatchObject({ role: "system", content: [{ type: "text", text: "Operator update." }] }) expect(decoded.messages[0]).toMatchObject({ role: "system", content: [{ type: "text", text: "Operator update." }] })
}) })
test("rejects chronological system updates between a local tool call and its result", async () => {
const messages = [
Message.assistant([
ToolCallPart.make({ id: "call_1", name: "lookup", input: {} }),
{ type: "text", text: "Waiting." },
]),
]
await expect(Effect.runPromise(ProviderShared.guardSystemUpdatePlacement("Test", messages, 1))).rejects.toThrow(
"Test chronological system updates cannot appear between a local tool call and its tool result",
)
})
test("rejects chronological system updates between results for multiple local tool calls", async () => {
const messages = [
Message.assistant([
ToolCallPart.make({ id: "call_1", name: "lookup", input: {} }),
ToolCallPart.make({ id: "call_2", name: "lookup", input: {} }),
]),
Message.tool({ id: "call_1", name: "lookup", result: "first" }),
]
await expect(Effect.runPromise(ProviderShared.guardSystemUpdatePlacement("Test", messages, 2))).rejects.toThrow(
"Test chronological system updates cannot appear between a local tool call and its tool result",
)
})
test("rejects invalid event type", () => { test("rejects invalid event type", () => {
expect(() => decodeLLMEvent({ type: "bogus" })).toThrow() expect(() => decodeLLMEvent({ type: "bogus" })).toThrow()
}) })

420
specs/v2/refreshable-context-sources.md
View file

@ -1,420 +0,0 @@
# Refreshable Context Sources
## Status
Reviewed proposal for ambient `AGENTS.md`, configured instruction paths or URLs, and later local skill-source invalidation.
## Decision Summary
Context-source observation remains pull-based and lazy at a safe provider-turn boundary.
```text
source signal
-> mark an optional observation cache stale
-> next naturally scheduled safe provider-turn boundary observes current state
-> Context Epoch compares and admits exact changed bytes durably
```
The first ambient `AGENTS.md` slice will not depend on filesystem watching. Its scoped contributor will directly observe local instruction state whenever `SystemContextRegistry.load()` naturally runs before a provider turn.
Watcher-backed caches are a later efficiency optimization for roots with proven subscription coverage. URLs remain separate observations with an independently chosen refresh policy.
## Existing Pieces
| Existing piece | Responsibility |
| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ |
| `Watcher.locationLayer` | Publish advisory `file.watcher.updated` events for local filesystem changes. |
| `EventV2.subscribe(...)` | Expose advisory events as scoped Effect streams. |
| `State.create(...)` | Rebuild replayable plugin and config contribution state from scoped transforms. |
| `SynchronizedRef.modifyEffect(...)` | Serialize effectful state refresh and store the next value only after success. |
| `SystemContext` | Convert coherent source samples into one immutable baseline, chronological updates, unavailable state, and removal tombstones. |
| `SystemContextRegistry` | Assemble Location-scoped built-in, instruction, and plugin context producers in stable contribution-key order. |
| `LocationServiceMap` | Own and clean up Location-scoped services, watcher subscriptions, and observation caches together. |
The missing reusable piece is deliberately small: retain the last successful value, mark it stale, and serialize refresh attempts.
## Primitive: `Refreshable`
Place the optional coordination helper at:
```text
packages/core/src/effect/refreshable.ts
```
`Refreshable` does not know about files, URLs, timers, watchers, Sessions, Context Epochs, or stale fallbacks. Domain services decide when to use it and how to recover expected observation failures.
```ts
export interface Refreshable<A, E = never, R = never> {
readonly get: Effect.Effect<A, E, R>
readonly invalidate: Effect.Effect<void>
}
export const make = <A, E, R>(
load: Effect.Effect<A, E, R>,
): Effect.Effect<Refreshable<A, E, R>>
```
Internal state:
```ts
type State<A> =
| { readonly _tag: "Empty" }
| { readonly _tag: "Fresh"; readonly value: A }
| { readonly _tag: "Stale"; readonly value: A }
```
Implementation substrate:
```text
SynchronizedRef.modifyEffect(...)
```
Semantics:
| Operation | Behavior |
| ---------------------------------- | ---------------------------------------------------------------------------------------------------------- |
| `get` on `Empty` | Run `load`, store `Fresh(value)` only after success, and return it. |
| `get` on `Fresh(value)` | Return the cached value without I/O. |
| `get` on `Stale(previous)` | Run `load`, store `Fresh(value)` only after success, and return it. |
| `invalidate` on `Fresh(value)` | Store `Stale(value)`. |
| `invalidate` on `Empty` or `Stale` | No-op. Repeated invalidations coalesce. |
| failed `load` | Preserve the prior `Empty` or `Stale(previous)` state, propagate the failure, and retry on the next `get`. |
| invalidation during `load` | Serialize after the reload and leave the refreshed cache stale for the next `get`. |
```mermaid
stateDiagram-v2
[*] --> Empty
Empty --> Fresh: get / load succeeds
Empty --> Empty: get / load fails
Fresh --> Fresh: get / return cached
Fresh --> Stale: invalidate
Stale --> Fresh: get / reload succeeds
Stale --> Stale: invalidate or reload fails
```
### Why Custom Instead Of Existing Effect Caches
Effect `Cache`, `ScopedCache`, and `Effect.cachedInvalidateWithTTL(...)` cache failed exits. Effect `Resource` preserves its prior value after a failed refresh, but eagerly acquires and does not reload lazily after explicit invalidation. Context-source observation requires the narrower lazy `Empty` / `Fresh` / `Stale` rule:
```text
failed refresh
-> retain prior successful value internally
-> remain stale
-> retry at the next natural request
```
`SynchronizedRef.modifyEffect(...)` commits the next state only after the refresh effect succeeds.
### Why No `peek`
`Refreshable` should not expose stale fallback reads. A domain service that needs a previous discovery graph during failed rescans should own that graph explicitly in its observation model.
### Why No Refresh Modes Or TTL
An always-refreshed source does not need a cache: load it directly. TTL expiry, watcher events, and explicit source changes are external invalidation policies that may call `invalidate` later.
```mermaid
sequenceDiagram
participant Signal as Optional Invalidation Source
participant Cache as Refreshable
participant Consumer
participant Loader
Signal->>Cache: invalidate
Consumer->>Cache: get
Cache->>Loader: reload only when stale
Loader-->>Cache: successful coherent value
Cache-->>Consumer: refreshed value
```
## Observation Units
Compose refreshables around coherent observations that share one invalidation policy. Do not create one uniformly per rendered Context Source or one aggregate cache for unrelated source kinds.
```text
local built-in discovery
-> one coherent observation while it shares one refresh policy
configured local glob
-> separate observation when its scan root or coverage differs
configured URL
-> independent observation
local skill directory
-> naturally one observation per registered source
embedded skill
-> direct value, no refreshable
```
## Ambient Instruction Contributor
Add a Location-scoped contributor to `SystemContextRegistry`:
```ts
yield *
registry.contribute({
key: SystemContext.Key.make("core/instructions"),
load: loadAmbientInstructions(),
})
```
`InstructionContext` owns instruction discovery, deterministic ordering, and source loading. `SystemContextRegistry` owns contributor composition and lifecycle. `SystemContext` remains unaware of files and URLs.
The first slice closes one coherent ordered instruction set into an aggregate source:
```text
core/instructions
-> [{ path, content }, ...]
```
Rendered text retains the human-readable source identity:
```text
Instructions from: /repo/packages/core/AGENTS.md
<exact file contents>
```
or:
```text
Instructions from: https://example.com/shared-agents.md
<exact response body>
```
The first implementation directly observes global and upward project `AGENTS.md` files on every safe provider-turn boundary. It does not use `Refreshable` yet unless a coherent source observation needs stale-on-failure retention.
```mermaid
sequenceDiagram
participant Runner as Safe Provider Boundary
participant Registry as System Context Registry
participant Instructions as Instruction Context
participant Files
participant Epoch as Context Epoch
Runner->>Registry: load
Registry->>Instructions: run contribution
Instructions->>Files: discover and read AGENTS.md files
Files-->>Instructions: coherent current observation
Instructions-->>Registry: instruction SystemContext
Registry-->>Runner: composed SystemContext
Runner->>Epoch: compare and durably admit changes
```
## Source Outcomes
Discovery and file reads form one coherent aggregate observation in the first slice.
```text
successful discovery and reads
-> one ordered aggregate instruction value
temporary discovery or read failure
-> aggregate SystemContext.unavailable
```
| Observation | Source outcome |
| -------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ |
| Local scan succeeds and discovers readable file | Include its exact contents in the available aggregate source. |
| Local scan succeeds and a previously discovered file is absent | Remove it from the aggregate value; remove the aggregate source when no instructions remain. |
| Local scan or file read fails transiently | Preserve the admitted aggregate source as `SystemContext.unavailable`; never emit mass removals. |
| Empty local file | Include the empty exact content in the available aggregate source. |
| URL returns `2xx` body | Available source with exact contents. |
| URL times out or returns transient failure | `SystemContext.unavailable`. |
| URL returns `404` or `410` | Decide the explicit removal contract before URL implementation. |
Aggregate instruction removal text must be model-meaningful:
```text
Previously loaded instructions no longer apply.
```
## First Ambient Slice
Implement only:
```text
global config AGENTS.md
+ upward project AGENTS.md ancestors
+ one aggregate core/instructions source
+ direct safe-turn observation
```
Preserve V1 ancestor stacking for `AGENTS.md`: nearest ancestor first, then outward through the project boundary.
Do not include yet:
```text
CLAUDE.md compatibility fallback
deprecated CONTEXT.md fallback
configured local paths or globs
configured URLs
watcher-backed caching
skills migration
nested read-triggered discovery
```
Tests:
```text
initial baseline
edit
newly added ancestor AGENTS.md
confirmed unlink with meaningful removal text
empty file
transient scan failure
transient file-read failure
deterministic ordering
restart with durable structured snapshots
```
## Future Watcher Optimization
Watchers remain advisory optimizations. They never wake idle Sessions and never publish durable Session context events.
The current watcher subscribes only to `location.directory`, has ignore rules, starts asynchronously, and swallows subscription failures. Effective source roots may live elsewhere. Do not expose one broad `Watcher.Service.local: "watching" | "poll"` flag.
Add root-specific watching only when needed:
```ts
export interface Watcher.Interface {
readonly watch: (root: AbsolutePath) => Effect.Effect<"watching" | "poll">
}
```
The exact watcher API remains a follow-up design. A truthful contract must account for:
- successful subscription startup;
- subscription failure and callback error;
- source roots above or outside `location.directory`;
- ignore and protected-path coverage;
- scope cleanup;
- own-process mutations that should synchronously invalidate caches before the next continuation turn.
When root coverage is proven:
```mermaid
sequenceDiagram
participant Watcher
participant Cache as Refreshable
participant Runner
participant Epoch as Context Epoch
Watcher->>Cache: relevant add / change / unlink
Cache->>Cache: invalidate only
Runner->>Cache: get at next safe boundary
Cache-->>Runner: refreshed coherent observation
Runner->>Epoch: durably admit exact changes
```
If coverage is not proven, bypass the cache and observe directly whenever the safe boundary naturally requests current state. This is safe-turn refresh, not a background polling loop.
When coverage is proven, cache each known candidate instruction path independently rather than invalidating one aggregate instruction cache:
```text
candidate instruction path
-> one Refreshable<File | Absent>
-> watcher event invalidates only the matching path
-> next safe provider boundary reloads only stale candidates
-> available candidates become ordered per-file Context Sources
```
Ambient candidates include the global `AGENTS.md` path and one `AGENTS.md` candidate in every applicable ancestor directory, including candidates that are currently absent so later additions are observable.
## URL Sources
URLs never share an observation cache with local discovery.
Start with direct safe-turn loading:
```text
safe provider-turn boundary
-> fetch URL
-> emit available or unavailable source
```
If measurements show excessive requests, add a URL-specific invalidation policy later:
```text
TTL expires
-> invalidate URL Refreshable
-> next safe provider-turn boundary reloads URL
```
A TTL timer must not wake idle Sessions or publish durable Session events.
## Skills Reuse
`SkillV2` currently stores replayable source registrations through `State.create(...)` and materialized source results in a raw permanent `Map<string, Info[]>`.
Use `Refreshable` later for local directory sources only after skill observation distinguishes confirmed absence from transient failure:
```text
State.create
-> current replayable Source registrations
private Map<Source.key, Refreshable>
-> reconcile against active non-embedded source keys
-> create refreshable for added source
-> drop refreshable for removed source
```
Do not add a generic `State -> cache invalidation` bridge. The skill service owns both current registrations and its observation-cache lifecycle.
Remote skill refresh remains a separate design slice because the current puller skips files that already exist and therefore does not define overwrite or removal semantics.
## Nested Instruction Discovery
Nested instructions discovered after successful read-tool activity remain a Session-scoped follow-up.
- The Location-scoped instruction service may resolve and reuse source observations.
- The set of nested source identities active for one Session must be durable and Session-scoped.
- Successful local file reads record newly observed nested source identities through synchronized Session events.
- The next safe provider boundary loads and admits them through Context Epoch history.
- Do not inject V1-style reminder text directly into read-tool output.
- Do not activate nested discovery for directory reads, managed tool-output resources, or external references until those semantics are explicitly designed.
## Lifecycle
- Location scope owns the System Context Registry, scoped context contributions, optional watcher-consumer fibers, and refreshable state.
- `Effect.forkScoped(...)` interrupts watcher-consumer fibers when the cached Location runtime is disposed.
- Stream finalization unsubscribes `EventV2` PubSub subscriptions.
- `Watcher.locationLayer` separately finalizes native Parcel watcher subscriptions.
- Repeated invalidations coalesce into one `Stale` state.
- Idle Sessions are not woken by local edits, URL timers, or plugin changes.
- Context Epoch admission remains serialized by the Session event transaction at the next naturally scheduled provider turn.
## Implementation Status And Follow-Up Order
Implemented in the direct-observation slice:
1. Add the Location-scoped `SystemContextRegistry` backed by stable-keyed scoped contributions.
2. Register built-in and ambient instruction producers with `SystemContextRegistry`.
3. Observe local instructions directly at each safe provider boundary.
4. Preserve admitted instructions after transient scan/read failure and block initial provider turns while context is unavailable.
5. Test ordering, edit, unlink, empty file, transient scan failure, discovered-then-missing races, durable restart behavior, and deterministic context admission.
Follow-up order:
1. Add and unit-test `Refreshable.make(load)` with `get` and `invalidate`.
2. Add truthful root-specific watcher registration.
3. Move ambient instructions from one directly observed aggregate to one watcher-invalidated Refreshable and Context Source per candidate file.
4. Add configured local exact paths and globs.
5. Add configured URL observations with explicit `404` and `410` semantics.
6. Migrate local `SkillV2` directory observations to per-source refreshables after skill failure semantics are corrected.
7. Add durable Session-scoped nested read discovery.
## Open Questions
1. Should configured URL sources treat `404` and `410` as confirmed removals?
2. What root-specific watcher API cleanly models ignore policy and callback health?
3. Should own-process file mutations publish an advisory invalidation event synchronously after commit?
## Compression Line
```text
SystemContextRegistry remembers which context producers participate.
Refreshable remembers whether a successful observation needs loading again.
Context Epoch remembers what the model was told.
```