mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-18 21:23:30 +00:00
fix(tui): reconcile session state after reconnect (#35262)
This commit is contained in:
parent
62af66a74f
commit
9daa4d85a4
4 changed files with 207 additions and 34 deletions
|
|
@ -104,6 +104,25 @@ bun dev api <operationId> --param key=value
|
|||
- If no compatible background server is registered, `bun dev api` starts one through the daemon service. Use `bun dev service status`, `bun dev service restart`, and `bun dev service stop` when you need explicit lifecycle control.
|
||||
- Prefer raw method/path calls for quick server debugging and operation IDs when exercising documented OpenAPI routes with path or query parameters.
|
||||
|
||||
## Auditing installed `opencode2` sessions
|
||||
|
||||
Installed next-channel sessions normally use `~/.local/share/opencode/opencode-next.db` and `~/.local/share/opencode/log/opencode.log`; `OPENCODE_DB` can override the database. Before calling `opencode2 api`, inspect `~/.local/state/opencode/service.json` because the command may start a daemon when none is healthy.
|
||||
|
||||
For a supplied `ses_...` ID, compare three sources:
|
||||
|
||||
- `opencode2 api get /api/session/active` and the Session/message endpoints for live server state.
|
||||
- The database's ordered `event` rows for durable history.
|
||||
- `packages/tui/src/context/data.tsx` and the relevant route for client projection and rendering.
|
||||
|
||||
Locate an uncertain database without modifying it:
|
||||
|
||||
```bash
|
||||
SESSION=ses_...
|
||||
for db in ~/.local/share/opencode/*.db; do
|
||||
sqlite3 "file:$db?mode=ro" "select 1 from session where id='$SESSION' limit 1" 2>/dev/null | grep -q 1 && printf '%s\n' "$db"
|
||||
done
|
||||
```
|
||||
|
||||
## Logs
|
||||
|
||||
- Log files live under `~/.local/share/opencode/log/`. In a local/dev checkout the active file is `opencode-local.log`; `opencode.log` is used for non-local (released) channel installs. Both are append-only, shared across every CLI and server process on the machine.
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ import type {
|
|||
SkillV2Info,
|
||||
V2Event,
|
||||
} from "@opencode-ai/sdk/v2"
|
||||
import { createStore, produce } from "solid-js/store"
|
||||
import { createStore, produce, reconcile } from "solid-js/store"
|
||||
import { createSimpleContext } from "./helper"
|
||||
import { useSDK } from "./sdk"
|
||||
import { createSignal, onCleanup } from "solid-js"
|
||||
|
|
@ -100,8 +100,15 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
directory: process.cwd(),
|
||||
})
|
||||
const messageIndex = new Map<string, Map<string, number>>()
|
||||
let connectionGeneration = 0
|
||||
let statusChanges: Set<string> | undefined
|
||||
let bootstrapping: Promise<void> | undefined
|
||||
|
||||
function setSessionStatus(sessionID: string, status: DataSessionStatus) {
|
||||
statusChanges?.add(sessionID)
|
||||
setStore("session", "status", sessionID, status)
|
||||
}
|
||||
|
||||
const message = {
|
||||
update(sessionID: string, fn: (messages: SessionMessage[], index: Map<string, number>) => void) {
|
||||
setStore(
|
||||
|
|
@ -260,16 +267,19 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
setStore("session", "info", event.data.sessionID, "title", event.data.title)
|
||||
break
|
||||
case "session.prompt.promoted": {
|
||||
setStore("session", "status", event.data.sessionID, "running")
|
||||
setSessionStatus(event.data.sessionID, "running")
|
||||
message.update(event.data.sessionID, (draft, index) => {
|
||||
const position = index.get(event.data.inputID)
|
||||
const existing = position === undefined ? undefined : draft[position]
|
||||
if (existing?.type === "user") {
|
||||
if (position === undefined) return
|
||||
const existing = draft[position]
|
||||
if (existing?.type === "user" && existing.metadata?.queued === true) {
|
||||
existing.time.created = event.created
|
||||
if (existing.metadata?.queued === true) {
|
||||
delete existing.metadata.queued
|
||||
if (Object.keys(existing.metadata).length === 0) existing.metadata = undefined
|
||||
}
|
||||
delete existing.metadata.queued
|
||||
if (Object.keys(existing.metadata).length === 0) existing.metadata = undefined
|
||||
draft.splice(position, 1)
|
||||
draft.push(existing)
|
||||
index.clear()
|
||||
draft.forEach((message, indexValue) => index.set(message.id, indexValue))
|
||||
return
|
||||
}
|
||||
})
|
||||
|
|
@ -311,7 +321,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
})
|
||||
break
|
||||
case "session.shell.started":
|
||||
setStore("session", "status", event.data.sessionID, "running")
|
||||
setSessionStatus(event.data.sessionID, "running")
|
||||
message.update(event.data.sessionID, (draft, index) => {
|
||||
message.append(draft, index, {
|
||||
id: messageIDFromEvent(event.id),
|
||||
|
|
@ -322,7 +332,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
})
|
||||
break
|
||||
case "session.shell.ended":
|
||||
setStore("session", "status", event.data.sessionID, "idle")
|
||||
setSessionStatus(event.data.sessionID, "idle")
|
||||
message.update(event.data.sessionID, (draft) => {
|
||||
const match = message.shell(draft, event.data.shell.id)
|
||||
if (!match) return
|
||||
|
|
@ -332,7 +342,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
})
|
||||
break
|
||||
case "session.step.started":
|
||||
setStore("session", "status", event.data.sessionID, "running")
|
||||
setSessionStatus(event.data.sessionID, "running")
|
||||
message.update(event.data.sessionID, (draft, index) => {
|
||||
if (index.has(event.data.assistantMessageID)) return
|
||||
const currentAssistant = message.activeAssistant(draft)
|
||||
|
|
@ -349,7 +359,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
})
|
||||
break
|
||||
case "session.step.ended":
|
||||
setStore("session", "status", event.data.sessionID, "running")
|
||||
setSessionStatus(event.data.sessionID, "running")
|
||||
message.update(event.data.sessionID, (draft, index) => {
|
||||
const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID)
|
||||
if (!currentAssistant) return
|
||||
|
|
@ -529,10 +539,10 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
break
|
||||
case "session.retried":
|
||||
case "session.compaction.started":
|
||||
setStore("session", "status", event.data.sessionID, "running")
|
||||
setSessionStatus(event.data.sessionID, "running")
|
||||
break
|
||||
case "session.execution.settled":
|
||||
setStore("session", "status", event.data.sessionID, "idle")
|
||||
setSessionStatus(event.data.sessionID, "idle")
|
||||
break
|
||||
case "session.revert.staged":
|
||||
if (store.session.info[event.data.sessionID])
|
||||
|
|
@ -869,15 +879,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
)
|
||||
for (const session of response.data) registerSession(session.id)
|
||||
}),
|
||||
sdk.api.session
|
||||
.active()
|
||||
.then((active) =>
|
||||
setStore(
|
||||
"session",
|
||||
"status",
|
||||
Object.fromEntries(Object.keys(active.data).map((sessionID) => [sessionID, "running" as const])),
|
||||
),
|
||||
),
|
||||
result.location.refresh(),
|
||||
result.location.agent.refresh(),
|
||||
result.location.integration.refresh(),
|
||||
|
|
@ -899,9 +900,30 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
|
|||
return bootstrapping
|
||||
}
|
||||
|
||||
function refreshActive() {
|
||||
const generation = ++connectionGeneration
|
||||
const changed = new Set<string>()
|
||||
statusChanges = changed
|
||||
void sdk.api.session
|
||||
.active()
|
||||
.then((active) => {
|
||||
if (generation !== connectionGeneration) return
|
||||
const status: Record<string, DataSessionStatus> = Object.fromEntries(
|
||||
Object.keys(active.data).map((sessionID) => [sessionID, "running" as const]),
|
||||
)
|
||||
for (const sessionID of changed) status[sessionID] = store.session.status[sessionID]
|
||||
setStore("session", "status", reconcile(status))
|
||||
})
|
||||
.catch(() => undefined)
|
||||
.finally(() => {
|
||||
if (statusChanges === changed) statusChanges = undefined
|
||||
})
|
||||
}
|
||||
|
||||
onCleanup(
|
||||
sdk.event.listen(({ details }) => {
|
||||
if (details.type === "server.connected") {
|
||||
refreshActive()
|
||||
void bootstrap()
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,23 +28,24 @@ export function createSessionRows(sessionID: Accessor<string>) {
|
|||
function reduce() {
|
||||
const messages = data.session.message.list(sessionID())
|
||||
const boundary = revertBoundary()
|
||||
return reduceSessionRows(boundary ? messages.filter((message) => message.id < boundary) : messages)
|
||||
const rows = reduceSessionRows(boundary ? messages.filter((message) => message.id < boundary) : messages)
|
||||
partitionPending(rows, pendingPermissions())
|
||||
return rows
|
||||
}
|
||||
|
||||
createEffect(() => {
|
||||
const pending = new Set(
|
||||
function pendingPermissions() {
|
||||
return new Set(
|
||||
(data.session.permission.list(sessionID()) ?? []).flatMap((request) =>
|
||||
request.source?.type === "tool" ? [request.source.callID] : [],
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
createEffect(() => {
|
||||
const pending = pendingPermissions()
|
||||
setRows(
|
||||
produce((draft) => {
|
||||
draft.forEach((row) => {
|
||||
if (row.type !== "group") return
|
||||
const refs = [...row.refs, ...row.pending]
|
||||
row.refs = refs.filter((ref) => !pending.has(ref.partID))
|
||||
row.pending = refs.filter((ref) => pending.has(ref.partID))
|
||||
})
|
||||
partitionPending(draft, pending)
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
|
@ -69,6 +70,20 @@ export function createSessionRows(sessionID: Accessor<string>) {
|
|||
}),
|
||||
)
|
||||
|
||||
createEffect(
|
||||
on(
|
||||
() =>
|
||||
data.session.message
|
||||
.list(sessionID())
|
||||
.flatMap((message) =>
|
||||
message.type === "user"
|
||||
? [{ id: message.id, created: message.time.created, queued: message.metadata?.queued === true }]
|
||||
: [],
|
||||
),
|
||||
() => setRows(reconcile(reduce())),
|
||||
),
|
||||
)
|
||||
|
||||
const appendMessage = (messageID: string) =>
|
||||
setRows(
|
||||
produce((draft) => {
|
||||
|
|
@ -134,7 +149,6 @@ export function createSessionRows(sessionID: Accessor<string>) {
|
|||
}
|
||||
const subscriptions = [
|
||||
data.on("session.prompt.admitted", input),
|
||||
data.on("session.prompt.promoted", input),
|
||||
data.on("session.context.updated", message),
|
||||
data.on("session.synthetic", (event) => {
|
||||
if (event.data.sessionID === sessionID() && event.data.description?.trim())
|
||||
|
|
@ -225,6 +239,15 @@ function completePrevious(rows: SessionRow[], index = rows.length) {
|
|||
if (previous?.type === "group") previous.completed = true
|
||||
}
|
||||
|
||||
function partitionPending(rows: SessionRow[], pending: Set<string>) {
|
||||
rows.forEach((row) => {
|
||||
if (row.type !== "group") return
|
||||
const refs = [...row.refs, ...row.pending]
|
||||
row.refs = refs.filter((ref) => !pending.has(ref.partID))
|
||||
row.pending = refs.filter((ref) => pending.has(ref.partID))
|
||||
})
|
||||
}
|
||||
|
||||
function exploration(name: string) {
|
||||
return ["read", "glob", "grep"].includes(name.toLowerCase())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import { onMount } from "solid-js"
|
|||
import { ProjectProvider } from "../../../src/context/project"
|
||||
import { SDKProvider } from "../../../src/context/sdk"
|
||||
import { DataProvider, useData } from "../../../src/context/data"
|
||||
import { createSessionRows } from "../../../src/routes/session/rows"
|
||||
import { createApi, createClient, createEventStream, createFetch, directory, json } from "../../fixture/tui-sdk"
|
||||
import { TestTuiContexts } from "../../fixture/tui-environment"
|
||||
|
||||
|
|
@ -109,12 +110,20 @@ test("refreshes resources into reactive getters", async () => {
|
|||
|
||||
test("reconnects the event stream and bootstraps fresh data", async () => {
|
||||
const events = createEventStream()
|
||||
const requests = { event: 0, model: 0 }
|
||||
const requests = { active: 0, event: 0, model: 0 }
|
||||
let resolveActive!: (response: Response) => void
|
||||
const calls = createFetch((url) => {
|
||||
if (url.pathname === "/api/event") {
|
||||
requests.event++
|
||||
return events.v2()
|
||||
}
|
||||
if (url.pathname === "/api/session/active") {
|
||||
requests.active++
|
||||
if (requests.active === 1) return json({ data: { "session-stale": { type: "running" } }, watermarks: {} })
|
||||
return new Promise<Response>((resolve) => {
|
||||
resolveActive = resolve
|
||||
})
|
||||
}
|
||||
if (url.pathname !== "/api/model") return
|
||||
requests.model++
|
||||
return json({
|
||||
|
|
@ -157,6 +166,7 @@ test("reconnects the event stream and bootstraps fresh data", async () => {
|
|||
|
||||
try {
|
||||
await wait(() => data.location.model.list()?.[0]?.id === "model-1")
|
||||
await wait(() => data.session.status("session-stale") === "running")
|
||||
expect(data.connection.status()).toBe("connected")
|
||||
expect(data.connection.attempt()).toBe(0)
|
||||
|
||||
|
|
@ -165,7 +175,25 @@ test("reconnects the event stream and bootstraps fresh data", async () => {
|
|||
expect(data.connection.attempt()).toBe(1)
|
||||
expect(data.connection.error()).toBe("Event stream disconnected")
|
||||
|
||||
await wait(() => requests.active === 2 && data.connection.status() === "connected", 4000)
|
||||
emitEvent(events, {
|
||||
id: "evt_step_started_after_reconnect",
|
||||
created: 1,
|
||||
type: "session.step.started",
|
||||
durable: durable("session-new"),
|
||||
data: {
|
||||
sessionID: "session-new",
|
||||
assistantMessageID: "message-new",
|
||||
agent: "build",
|
||||
model: { id: "model", providerID: "provider" },
|
||||
},
|
||||
})
|
||||
await wait(() => data.session.status("session-new") === "running")
|
||||
resolveActive(json({ data: {}, watermarks: {} }))
|
||||
|
||||
await wait(() => data.location.model.list()?.[0]?.id === "model-2", 4000)
|
||||
await wait(() => data.session.status("session-stale") === "idle")
|
||||
expect(data.session.status("session-new")).toBe("running")
|
||||
expect(requests.event).toBe(2)
|
||||
expect(data.connection.status()).toBe("connected")
|
||||
expect(data.connection.attempt()).toBe(0)
|
||||
|
|
@ -175,6 +203,87 @@ test("reconnects the event stream and bootstraps fresh data", async () => {
|
|||
}
|
||||
})
|
||||
|
||||
test("completes exploration when a queued prompt is promoted", async () => {
|
||||
const events = createEventStream()
|
||||
const sessionID = "session-promotion"
|
||||
const calls = createFetch((url) => {
|
||||
if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} })
|
||||
}, events)
|
||||
let rows!: ReturnType<typeof createSessionRows>
|
||||
|
||||
function Probe() {
|
||||
rows = createSessionRows(() => sessionID)
|
||||
return <box />
|
||||
}
|
||||
|
||||
const app = await testRender(() => (
|
||||
<TestTuiContexts>
|
||||
<SDKProvider client={createClient(calls.fetch)} api={createApi(calls.fetch)}>
|
||||
<ProjectProvider>
|
||||
<DataProvider>
|
||||
<Probe />
|
||||
</DataProvider>
|
||||
</ProjectProvider>
|
||||
</SDKProvider>
|
||||
</TestTuiContexts>
|
||||
))
|
||||
|
||||
try {
|
||||
emitEvent(events, {
|
||||
id: "evt_step_started",
|
||||
created: 1,
|
||||
type: "session.step.started",
|
||||
durable: durable(sessionID),
|
||||
data: {
|
||||
sessionID,
|
||||
assistantMessageID: "message-assistant",
|
||||
agent: "build",
|
||||
model: { id: "model", providerID: "provider" },
|
||||
},
|
||||
})
|
||||
emitEvent(events, {
|
||||
id: "evt_tool_started",
|
||||
created: 2,
|
||||
type: "session.tool.input.started",
|
||||
durable: durable(sessionID, 1),
|
||||
data: {
|
||||
sessionID,
|
||||
assistantMessageID: "message-assistant",
|
||||
callID: "call-read",
|
||||
name: "read",
|
||||
},
|
||||
})
|
||||
await wait(() => rows.some((row) => row.type === "group" && !row.completed))
|
||||
|
||||
emitEvent(events, {
|
||||
id: "evt_prompt_admitted",
|
||||
created: 3,
|
||||
type: "session.prompt.admitted",
|
||||
durable: durable(sessionID, 2),
|
||||
data: {
|
||||
sessionID,
|
||||
inputID: "message-user",
|
||||
prompt: { text: "Continue" },
|
||||
delivery: "steer",
|
||||
},
|
||||
})
|
||||
await wait(() => rows.at(-1)?.type === "message")
|
||||
expect(rows.find((row) => row.type === "group")?.completed).toBe(false)
|
||||
|
||||
emitEvent(events, {
|
||||
id: "evt_prompt_promoted",
|
||||
created: 4,
|
||||
type: "session.prompt.promoted",
|
||||
durable: durable(sessionID, 3),
|
||||
data: { sessionID, inputID: "message-user" },
|
||||
})
|
||||
await wait(() => rows.find((row) => row.type === "group")?.completed === true)
|
||||
expect(rows.at(-1)).toEqual({ type: "message", messageID: "message-user" })
|
||||
} finally {
|
||||
app.renderer.destroy()
|
||||
}
|
||||
})
|
||||
|
||||
test("connectedOnce is false until first connect and persists across disconnect", async () => {
|
||||
const encoder = new TextEncoder()
|
||||
let stream: ReadableStreamDefaultController<Uint8Array> | undefined
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue