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

refactor(core): separate v2 message identity

Browse source
This commit is contained in:
Kit Langton 2026-06-04 12:08:00 -04:00
commit f2d133edf7
26 changed files with 772 additions and 289 deletions
Download patch file Download diff file Expand all files Collapse all files

98
packages/core/test/database-migration.test.ts
View file

@ -7,9 +7,11 @@ import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
import { Effect, Layer } from "effect"
import { eq, inArray, sql } from "drizzle-orm"
import { DatabaseMigration } from "@opencode-ai/core/database/migration"
import { migrations } from "@opencode-ai/core/database/migration.gen"
import sessionUsageMigration from "@opencode-ai/core/database/migration/20260510033149_session_usage"
import normalizeStoragePathsMigration from "@opencode-ai/core/database/migration/20260601010001_normalize_storage_paths"
import sessionMessageProjectionOrderMigration from "@opencode-ai/core/database/migration/20260603040000_session_message_projection_order"
import sessionMessageIdentityMigration from "@opencode-ai/core/database/migration/20260604153000_session_message_identity"
import { ProjectV2 } from "@opencode-ai/core/project"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import { AbsolutePath } from "@opencode-ai/core/schema"
@ -62,7 +64,7 @@ describe("DatabaseMigration", () => {
expect(
yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`),
).toEqual({ name: "session_input" })
expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: 29 })
expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: 30 })
expect(
yield* db.all(
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
@ -134,6 +136,100 @@ describe("DatabaseMigration", () => {
)
})
test("resets unreleased EventV2 state without deleting canonical Session history", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
yield* db.run(sql`CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL)`)
yield* db.run(sql`CREATE TABLE part (id text PRIMARY KEY, message_id text NOT NULL, session_id text NOT NULL)`)
yield* db.run(sql`CREATE TABLE session_message (id text PRIMARY KEY)`)
yield* db.run(sql`CREATE TABLE session_input (id text PRIMARY KEY)`)
yield* db.run(
sql`CREATE TABLE event_sequence (aggregate_id text PRIMARY KEY, seq integer NOT NULL, owner_id text)`,
)
yield* db.run(sql`CREATE TABLE event (id text PRIMARY KEY, aggregate_id text NOT NULL, seq integer NOT NULL)`)
yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
yield* db.run(sql`INSERT INTO message (id, session_id) VALUES ('legacy_message', 'session')`)
yield* db.run(
sql`INSERT INTO part (id, message_id, session_id) VALUES ('legacy_part', 'legacy_message', 'session')`,
)
yield* db.run(sql`INSERT INTO session_message (id) VALUES ('experimental_message')`)
yield* db.run(sql`INSERT INTO session_input (id) VALUES ('experimental_input')`)
yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq, owner_id) VALUES ('session', 1, 'workspace')`)
yield* db.run(sql`INSERT INTO event (id, aggregate_id, seq) VALUES ('experimental_event', 'session', 1)`)
yield* DatabaseMigration.applyOnly(db, [sessionMessageIdentityMigration])
expect(yield* db.all(sql`SELECT id FROM session`)).toEqual([{ id: "session" }])
expect(yield* db.all(sql`SELECT id FROM message`)).toEqual([{ id: "legacy_message" }])
expect(yield* db.all(sql`SELECT id FROM part`)).toEqual([{ id: "legacy_part" }])
expect(yield* db.all(sql`SELECT id FROM session_message`)).toEqual([])
expect(yield* db.all(sql`SELECT id FROM session_input`)).toEqual([])
expect(yield* db.all(sql`SELECT id FROM event`)).toEqual([])
expect(yield* db.all(sql`SELECT aggregate_id FROM event_sequence`)).toEqual([])
}),
)
})
test("applies the Session-message identity reset through normal database startup", async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "identity-reset.sqlite")
const before = migrations.filter((migration) => migration.id !== sessionMessageIdentityMigration.id)
await Effect.runPromise(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`PRAGMA foreign_keys = ON`)
yield* DatabaseMigration.applyOnly(db, before)
yield* db.run(
sql`INSERT INTO project (id, worktree, sandboxes, time_created, time_updated) VALUES ('project', '/project', '[]', 1, 1)`,
)
yield* db.run(
sql`INSERT INTO session (id, project_id, slug, directory, title, version, time_created, time_updated) VALUES ('session', 'project', 'session', '/project', 'Session', 'test', 1, 1)`,
)
yield* db.run(
sql`INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES ('legacy_message', 'session', 1, 1, '{}')`,
)
yield* db.run(
sql`INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES ('legacy_part', 'legacy_message', 'session', 1, 1, '{}')`,
)
yield* db.run(
sql`INSERT INTO todo (session_id, content, status, priority, position, time_created, time_updated) VALUES ('session', 'keep', 'pending', 'low', 0, 1, 1)`,
)
yield* db.run(
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('evt_message', 'session', 'user', 1, 1, 1, '{}')`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, prompt, delivery, time_created) VALUES ('evt_input', 'session', '{}', 'queue', 1)`,
)
yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq, owner_id) VALUES ('session', 1, 'workspace')`)
yield* db.run(
sql`INSERT INTO event (id, aggregate_id, seq, type, data) VALUES ('evt_event', 'session', 1, 'session.created.1', '{}')`,
)
}).pipe(Effect.provide(SqliteClient.layer({ filename, disableWAL: true })), Effect.scoped),
)
await Effect.runPromise(Effect.scoped(Layer.build(Database.layerFromPath(filename))))
await Effect.runPromise(
Effect.gen(function* () {
const db = yield* makeDb
expect(yield* db.all(sql`SELECT id FROM session`)).toEqual([{ id: "session" }])
expect(yield* db.all(sql`SELECT id FROM message`)).toEqual([{ id: "legacy_message" }])
expect(yield* db.all(sql`SELECT id FROM part`)).toEqual([{ id: "legacy_part" }])
expect(yield* db.all(sql`SELECT content FROM todo`)).toEqual([{ content: "keep" }])
expect(yield* db.all(sql`SELECT id FROM session_message`)).toEqual([])
expect(yield* db.all(sql`SELECT id FROM session_input`)).toEqual([])
expect(yield* db.all(sql`SELECT id FROM event`)).toEqual([])
expect(yield* db.all(sql`SELECT aggregate_id FROM event_sequence`)).toEqual([])
expect(yield* db.get(sql`SELECT id FROM migration WHERE id = ${sessionMessageIdentityMigration.id}`)).toEqual({
id: sessionMessageIdentityMigration.id,
})
}).pipe(Effect.provide(SqliteClient.layer({ filename, disableWAL: true })), Effect.scoped),
)
})
test("runs session usage backfill in order with schema changes", async () => {
await run(
Effect.gen(function* () {

16
packages/core/test/event.test.ts
View file

@ -7,6 +7,7 @@ import { Location } from "@opencode-ai/core/location"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { WorkspaceV2 } from "@opencode-ai/core/workspace"
import { V2Schema } from "@opencode-ai/core/v2-schema"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { eq } from "drizzle-orm"
import { location } from "./fixture/location"
import { testEffect } from "./lib/effect"
@ -84,6 +85,21 @@ const SyncTimestamp = EventV2.define({
})
describe("EventV2", () => {
it.effect("keeps event IDs in the evt namespace", () =>
Effect.sync(() => {
expect(EventV2.ID.create()).toMatch(/^evt_/)
expect(() => EventV2.ID.make("msg_wrong_namespace")).toThrow()
expect(() => EventV2.ID.make("evtx")).toThrow()
}),
)
it.effect("round-trips Session message IDs through creator event IDs", () =>
Effect.sync(() => {
expect(String(SessionMessage.ID.fromEvent(EventV2.ID.make("evt_custom")))).toBe("msg_custom")
expect(String(SessionMessage.ID.toEvent(SessionMessage.ID.make("msg_custom")))).toBe("evt_custom")
}),
)
it.effect("derives stable namespaced external IDs", () =>
Effect.sync(() => {
const input = { namespace: "opencord.agent-input", key: "input-1" }

123
packages/core/test/session-projector.test.ts
View file

@ -67,13 +67,23 @@ describe("SessionProjector", () => {
yield* events.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: created, prompt: new Prompt({ text: "first" }), delivery: "steer" },
{ id: SessionMessage.ID.make("evt_z") },
{
sessionID,
timestamp: created,
prompt: new Prompt({ text: "first" }),
delivery: "steer",
},
{ id: EventV2.ID.make("evt_z") },
)
yield* events.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: created, prompt: new Prompt({ text: "second" }), delivery: "steer" },
{ id: SessionMessage.ID.make("evt_a") },
{
sessionID,
timestamp: created,
prompt: new Prompt({ text: "second" }),
delivery: "steer",
},
{ id: EventV2.ID.make("evt_a") },
)
const sessions = yield* SessionV2.Service
@ -131,13 +141,13 @@ describe("SessionProjector", () => {
.run()
.pipe(Effect.orDie)
const events = yield* EventV2.Service
const id = SessionMessage.ID.make("evt_admitted")
const id = SessionMessage.ID.make("msg_admitted")
yield* SessionInput.admit(db, { id, sessionID, prompt: new Prompt({ text: "promote me" }), delivery: "steer" })
const event = yield* events.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: created, prompt: new Prompt({ text: "promote me" }), delivery: "steer" },
{ id },
{ id: SessionMessage.ID.toEvent(id) },
)
expect(
@ -168,9 +178,21 @@ describe("SessionProjector", () => {
.pipe(Effect.orDie)
const events = yield* EventV2.Service
yield* events.publish(SessionEvent.AgentSwitched, { sessionID, timestamp: created, agent: "build" })
yield* events.publish(SessionEvent.ModelSwitched, { sessionID, timestamp: created, model })
yield* events.publish(SessionEvent.Synthetic, { sessionID, timestamp: created, text: "synthetic context" })
yield* events.publish(SessionEvent.AgentSwitched, {
sessionID,
timestamp: created,
agent: "build",
})
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
timestamp: created,
model,
})
yield* events.publish(SessionEvent.Synthetic, {
sessionID,
timestamp: created,
text: "synthetic context",
})
yield* events.publish(SessionEvent.Shell.Started, {
sessionID,
timestamp: created,
@ -183,7 +205,11 @@ describe("SessionProjector", () => {
callID: "shell-1",
output: "/project",
})
yield* events.publish(SessionEvent.Compaction.Started, { sessionID, timestamp: created, reason: "manual" })
yield* events.publish(SessionEvent.Compaction.Started, {
sessionID,
timestamp: created,
reason: "manual",
})
yield* events.publish(SessionEvent.Compaction.Delta, { sessionID, timestamp: created, text: "partial" })
yield* events.publish(SessionEvent.Compaction.Ended, {
sessionID,
@ -228,6 +254,53 @@ describe("SessionProjector", () => {
}),
)
it.effect("rejects a creator event that reuses an existing projected message ID", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
.run()
.pipe(Effect.orDie)
yield* db
.insert(SessionTable)
.values({
id: sessionID,
project_id: Project.ID.global,
slug: "test",
directory: "/project",
title: "test",
version: "test",
})
.run()
.pipe(Effect.orDie)
const events = yield* EventV2.Service
yield* events.publish(
SessionEvent.Synthetic,
{ sessionID, timestamp: created, text: "first" },
{ id: EventV2.ID.make("evt_same") },
)
const duplicate = yield* events
.publish(
SessionEvent.Synthetic,
{ sessionID, timestamp: created, text: "second" },
{ id: EventV2.ID.make("evt_same") },
)
.pipe(Effect.exit)
expect(duplicate._tag).toBe("Failure")
expect(
yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, SessionMessage.ID.make("msg_same")))
.get()
.pipe(Effect.orDie),
).toMatchObject({ data: { text: "first" } })
}),
)
it.effect("rejects a Prompted event that conflicts with an admitted inbox row", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
@ -249,14 +322,14 @@ describe("SessionProjector", () => {
.run()
.pipe(Effect.orDie)
const events = yield* EventV2.Service
const id = SessionMessage.ID.make("evt_conflict")
const id = SessionMessage.ID.make("msg_conflict")
yield* SessionInput.admit(db, { id, sessionID, prompt: new Prompt({ text: "admitted" }), delivery: "steer" })
const exit = yield* events
.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: created, prompt: new Prompt({ text: "different" }), delivery: "steer" },
{ id },
{ id: SessionMessage.ID.toEvent(id) },
)
.pipe(Effect.exit)
@ -288,12 +361,16 @@ describe("SessionProjector", () => {
.run()
.pipe(Effect.orDie)
const events = yield* EventV2.Service
const id = SessionMessage.ID.make("evt_delivery_conflict")
const id = SessionMessage.ID.make("msg_delivery_conflict")
const prompt = new Prompt({ text: "admitted" })
yield* SessionInput.admit(db, { id, sessionID, prompt, delivery: "queue" })
const exit = yield* events
.publish(SessionEvent.Prompted, { sessionID, timestamp: created, prompt, delivery: "steer" }, { id })
.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: created, prompt, delivery: "steer" },
{ id: SessionMessage.ID.toEvent(id) },
)
.pipe(Effect.exit)
expect(String(exit)).toContain("Prompt projection conflicts with admitted input")
@ -306,7 +383,7 @@ describe("SessionProjector", () => {
it.effect("does not revive a stale incomplete in-memory assistant projection", () =>
Effect.gen(function* () {
const stale = new SessionMessage.Assistant({
id: SessionMessage.ID.make("evt_assistant_stale"),
id: SessionMessage.ID.make("msg_assistant_stale"),
type: "assistant",
agent: "build",
model,
@ -314,7 +391,7 @@ describe("SessionProjector", () => {
time: { created },
})
const completed = new SessionMessage.Assistant({
id: SessionMessage.ID.make("evt_assistant_completed"),
id: SessionMessage.ID.make("msg_assistant_completed"),
type: "assistant",
agent: "build",
model,
@ -351,8 +428,8 @@ describe("SessionProjector", () => {
yield* db
.insert(SessionMessageTable)
.values([
assistantRow(SessionMessage.ID.make("evt_assistant_1"), 0),
assistantRow(SessionMessage.ID.make("evt_assistant_2"), 1),
assistantRow(SessionMessage.ID.make("msg_assistant_1"), 0),
assistantRow(SessionMessage.ID.make("msg_assistant_2"), 1),
])
.run()
.pipe(Effect.orDie)
@ -361,7 +438,7 @@ describe("SessionProjector", () => {
yield* service.publish(SessionEvent.Step.Ended, {
sessionID,
timestamp: DateTime.makeUnsafe(1),
assistantMessageID: SessionMessage.ID.make("evt_assistant_2"),
assistantMessageID: EventV2.ID.make("evt_assistant_2"),
finish: "stop",
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
@ -409,8 +486,8 @@ describe("SessionProjector", () => {
yield* db
.insert(SessionMessageTable)
.values([
assistantRow(SessionMessage.ID.make("evt_assistant_stale"), 0),
assistantRow(SessionMessage.ID.make("evt_assistant_completed"), 1, {
assistantRow(SessionMessage.ID.make("msg_assistant_stale"), 0),
assistantRow(SessionMessage.ID.make("msg_assistant_completed"), 1, {
created: DateTime.makeUnsafe(1),
completed: DateTime.makeUnsafe(2),
}),
@ -437,7 +514,7 @@ describe("SessionProjector", () => {
)
expect(messages).toEqual([
new SessionMessage.Assistant({
id: SessionMessage.ID.make("evt_assistant_completed"),
id: SessionMessage.ID.make("msg_assistant_completed"),
type: "assistant",
agent: "build",
model,
@ -445,7 +522,7 @@ describe("SessionProjector", () => {
time: { created: DateTime.makeUnsafe(1), completed: DateTime.makeUnsafe(2) },
}),
new SessionMessage.Assistant({
id: SessionMessage.ID.make("evt_assistant_stale"),
id: SessionMessage.ID.make("msg_assistant_stale"),
type: "assistant",
agent: "build",
model,

53
packages/core/test/session-prompt.test.ts
View file

@ -310,7 +310,7 @@ describe("SessionV2.prompt", () => {
yield* events.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: yield* DateTime.now, prompt, delivery: "steer" },
{ id: messageID },
{ id: SessionMessage.ID.toEvent(messageID) },
)
const retried = yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
@ -329,7 +329,7 @@ describe("SessionV2.prompt", () => {
yield* events.publish(
SessionEvent.Prompted,
{ sessionID, timestamp: yield* DateTime.now, prompt, delivery: "queue" },
{ id: messageID },
{ id: SessionMessage.ID.toEvent(messageID) },
)
const retried = yield* session.prompt({ id: messageID, sessionID, prompt, delivery: "queue", resume: false })
@ -339,7 +339,7 @@ describe("SessionV2.prompt", () => {
}),
)
it.effect("rejects an input ID already used by a durable non-prompt event", () =>
it.effect("rejects an input ID already used by a projected non-prompt message", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
@ -347,7 +347,7 @@ describe("SessionV2.prompt", () => {
yield* events.publish(
SessionEvent.Synthetic,
{ sessionID, timestamp: yield* DateTime.now, text: "Collision" },
{ id: messageID },
{ id: SessionMessage.ID.toEvent(messageID) },
)
const failure = yield* session
@ -359,7 +359,7 @@ describe("SessionV2.prompt", () => {
}),
)
it.effect("rejects a durable event ID reserved by an admitted prompt without poisoning promotion", () =>
it.effect("keeps event envelope IDs separate from admitted message IDs", () =>
Effect.gen(function* () {
yield* setup
const { db } = yield* Database.Service
@ -368,24 +368,51 @@ describe("SessionV2.prompt", () => {
const prompt = new Prompt({ text: "Reserved prompt" })
yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
yield* events.publish(
SessionEvent.Synthetic,
{ sessionID, timestamp: yield* DateTime.now, text: "Synthetic" },
{ id: EventV2.ID.make("evt_reserved_prompt") },
)
expect(yield* admitted(messageID)).not.toHaveProperty("promotedSeq")
yield* SessionInput.promoteSteers(db, events, sessionID)
expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 1 })
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: messageID, type: "user", text: "Reserved prompt" },
{ type: "synthetic", text: "Synthetic" },
])
}),
)
it.effect("keeps Session message IDs in the msg namespace", () =>
Effect.sync(() => {
expect(SessionMessage.ID.create()).toMatch(/^msg_/)
expect(() => SessionMessage.ID.make("evt_wrong_namespace")).toThrow()
expect(() => SessionMessage.ID.make("msgx")).toThrow()
}),
)
it.effect("rejects a non-prompt event that reuses an admitted message ID", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const prompt = new Prompt({ text: "Reserved prompt" })
yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
const failure = yield* events
.publish(
SessionEvent.Synthetic,
{ sessionID, timestamp: yield* DateTime.now, text: "Conflicting synthetic" },
{ id: messageID },
{ id: SessionMessage.ID.toEvent(messageID) },
)
.pipe(Effect.catchDefect(Effect.succeed))
expect(failure).toBe("Durable event conflicts with admitted prompt input")
expect(yield* admitted(messageID)).not.toHaveProperty("promotedSeq")
expect(yield* session.messages({ sessionID })).toEqual([])
yield* SessionInput.promoteSteers(db, events, sessionID)
expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 0 })
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: messageID, type: "user", text: "Reserved prompt" },
])
}),
)

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

@ -1,7 +1,6 @@
import { describe, expect, test } from "bun:test"
import { Message, Model } from "@opencode-ai/llm"
import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat"
import { EventV2 } from "@opencode-ai/core/event"
import { ModelV2 } from "@opencode-ai/core/model"
import { ProviderV2 } from "@opencode-ai/core/provider"
import { SessionMessage } from "@opencode-ai/core/session/message"
@ -12,7 +11,7 @@ import { ToolOutput } from "@opencode-ai/core/tool-output"
import { DateTime } from "effect"
const created = DateTime.makeUnsafe(0)
const id = (value: string) => EventV2.ID.make(`evt_${value}`)
const id = (value: string) => SessionMessage.ID.make(`msg_${value}`)
const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.route })
describe("toLLMMessages", () => {

26
packages/core/test/session-runner.test.ts
View file

@ -1218,30 +1218,30 @@ describe("SessionRunnerLLM", () => {
const events = yield* EventV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Recover interrupted tool" }), resume: false })
yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
const assistant = yield* events.publish(SessionEvent.Step.Started, {
const assistantMessageID = (yield* events.publish(SessionEvent.Step.Started, {
sessionID,
timestamp: yield* DateTime.now,
agent: "build",
model: { id: ModelV2.ID.make("fake-model"), providerID: ProviderV2.ID.make("fake") },
})
})).id
yield* events.publish(SessionEvent.Tool.Input.Started, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-interrupted",
name: "echo",
})
yield* events.publish(SessionEvent.Tool.Input.Ended, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-interrupted",
text: '{"text":"stale"}',
})
yield* events.publish(SessionEvent.Tool.Called, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-interrupted",
tool: "echo",
input: { text: "stale" },
@ -1280,30 +1280,30 @@ describe("SessionRunnerLLM", () => {
resume: false,
})
yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
const assistant = yield* events.publish(SessionEvent.Step.Started, {
const assistantMessageID = (yield* events.publish(SessionEvent.Step.Started, {
sessionID,
timestamp: yield* DateTime.now,
agent: "build",
model: { id: ModelV2.ID.make("fake-model"), providerID: ProviderV2.ID.make("fake") },
})
})).id
yield* events.publish(SessionEvent.Tool.Input.Started, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-hosted-interrupted",
name: "web_search",
})
yield* events.publish(SessionEvent.Tool.Input.Ended, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-hosted-interrupted",
text: '{"query":"stale"}',
})
yield* events.publish(SessionEvent.Tool.Called, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-hosted-interrupted",
tool: "web_search",
input: { query: "stale" },
@ -1338,16 +1338,16 @@ describe("SessionRunnerLLM", () => {
resume: false,
})
yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
const assistant = yield* events.publish(SessionEvent.Step.Started, {
const assistantMessageID = (yield* events.publish(SessionEvent.Step.Started, {
sessionID,
timestamp: yield* DateTime.now,
agent: "build",
model: { id: ModelV2.ID.make("fake-model"), providerID: ProviderV2.ID.make("fake") },
})
})).id
yield* events.publish(SessionEvent.Tool.Input.Started, {
sessionID,
timestamp: yield* DateTime.now,
assistantMessageID: assistant.id,
assistantMessageID,
callID: "call-pending-interrupted",
name: "echo",
})

2
packages/core/test/session-tool-progress.test.ts
View file

@ -60,7 +60,7 @@ describe("Tool.Progress", () => {
const row = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, assistantMessageID))
.where(eq(SessionMessageTable.id, SessionMessage.ID.fromEvent(assistantMessageID)))
.get()
.pipe(Effect.orDie)
if (!row) return yield* Effect.die("Missing projected assistant")