diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 49000550db..84a14dbdb4 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -7,6 +7,7 @@ import { Bus } from "../bus" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { Model } from "../model" import { SessionEvent } from "./event" +import type { SessionSchema } from "./schema" import { SessionV1 } from "../v1/session" import { WorkspaceTable } from "../control-plane/workspace.sql" import { SessionMessage } from "./message" @@ -447,12 +448,33 @@ function run(db: DatabaseService, event: MessageEvent) { function runAndTouch(db: DatabaseService, event: MessageEvent) { return Effect.gen(function* () { yield* run(db, event) + yield* touchAncestors(db, event.data.sessionID, DateTime.toEpochMillis(event.created)) + }) +} + +function touchAncestors( + db: DatabaseService, + sessionID: SessionSchema.ID, + updated: number, + seen = new Set(), +): Effect.Effect { + if (seen.has(sessionID)) return Effect.void + seen.add(sessionID) + return Effect.gen(function* () { + const session = yield* db + .select({ parentID: SessionTable.parent_id }) + .from(SessionTable) + .where(eq(SessionTable.id, sessionID)) + .get() + .pipe(Effect.orDie) + if (!session) return yield* db .update(SessionTable) - .set({ time_updated: DateTime.toEpochMillis(event.created) }) - .where(eq(SessionTable.id, event.data.sessionID)) + .set({ time_updated: updated }) + .where(and(eq(SessionTable.id, sessionID), lt(SessionTable.time_updated, updated))) .run() .pipe(Effect.orDie) + if (session.parentID) yield* touchAncestors(db, session.parentID, updated, seen) }) } diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index f2b898c442..1395493c1d 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { DateTime, Effect, Fiber, Option, Schema, Stream } from "effect" -import { asc, eq, sql } from "drizzle-orm" +import { asc, eq, inArray, sql } from "drizzle-orm" import { Database } from "@opencode-ai/core/database/database" import { Agent } from "@opencode-ai/core/agent" import { LayerNode } from "@opencode-ai/util/effect/layer-node" @@ -659,7 +659,6 @@ describe("SessionProjector", () => { directory: "/project", title: "test", version: "test", - time_updated: -1, }) .run() .pipe(Effect.orDie) @@ -671,23 +670,63 @@ describe("SessionProjector", () => { .where(eq(SessionTable.id, sessionID)) .get() .pipe(Effect.orDie) - const updated = () => - db - .select({ value: SessionTable.time_updated }) - .from(SessionTable) - .where(eq(SessionTable.id, sessionID)) - .get() - .pipe(Effect.orDie) yield* bus.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: "shutdown" }) expect((yield* suspended())?.timeSuspended).toBeNull() - expect((yield* updated())?.value ?? -1).toBeGreaterThan(-1) yield* bus.publish(SessionEvent.Execution.Started, { sessionID }) expect((yield* suspended())?.timeSuspended).toBeNull() }), ) + it.effect("rolls terminal execution activity through every session ancestor", () => + 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) + const root = Session.ID.make("ses_activity_root") + const child = Session.ID.make("ses_activity_child") + const grandchild = Session.ID.make("ses_activity_grandchild") + yield* db + .insert(SessionTable) + .values( + [ + { id: root }, + { id: child, parent_id: root }, + { id: grandchild, parent_id: child }, + ].map((session) => ({ + ...session, + project_id: Project.ID.global, + slug: session.id, + directory: "/project", + title: "test", + version: "test", + time_updated: -1, + })), + ) + .run() + .pipe(Effect.orDie) + + yield* (yield* Bus.Service).publish(SessionEvent.Execution.Interrupted, { + sessionID: grandchild, + reason: "shutdown", + }) + + const rows = yield* db + .select({ id: SessionTable.id, updated: SessionTable.time_updated }) + .from(SessionTable) + .where(inArray(SessionTable.id, [root, child, grandchild])) + .all() + .pipe(Effect.orDie) + expect(rows).toHaveLength(3) + expect(rows.every((row) => row.updated > -1)).toBe(true) + expect(new Set(rows.map((row) => row.updated)).size).toBe(1) + }), + ) + it.effect("updates only the newest incomplete assistant projection", () => Effect.gen(function* () { const { db } = yield* Database.Service