Skip to content

fix(acp): persist session state to disk and add isReplay metadata #6563

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
myInstagramAlternative wants to merge 1 commit into
base: dev
Choose a base branch
from
+104 −17
Open

fix(acp): persist session state to disk and add isReplay metadata #6563

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 16 additions & 9 deletions packages/opencode/src/acp/agent.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -449,7 +449,7 @@ export namespace ACP {

for (const msg of messages ?? []) {
log.debug("replay message", msg)
await this.processMessage(msg)
await this.processMessage(msg, true)
}

return mode
Expand All @@ -464,7 +464,7 @@ export namespace ACP {
}
}

private async processMessage(message: SessionMessageResponse) {
private async processMessage(message: SessionMessageResponse, isReplay: boolean = false) {
log.debug("process message", message)
if (message.info.role !== "assistant" && message.info.role !== "user") return
const sessionId = message.info.sessionID
Expand All @@ -484,6 +484,7 @@ export namespace ACP {
status: "pending",
locations: [],
rawInput: {},
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand All @@ -500,6 +501,7 @@ export namespace ACP {
status: "in_progress",
locations: toLocations(part.tool, part.state.input),
rawInput: part.state.input,
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand Down Expand Up @@ -553,6 +555,7 @@ export namespace ACP {
content: todo.content,
}
}),
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand All @@ -577,6 +580,7 @@ export namespace ACP {
output: part.state.output,
metadata: part.state.metadata,
},
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand All @@ -603,6 +607,7 @@ export namespace ACP {
rawOutput: {
error: part.state.error,
},
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand All @@ -621,6 +626,7 @@ export namespace ACP {
type: "text",
text: part.text,
},
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand All @@ -638,6 +644,7 @@ export namespace ACP {
type: "text",
text: part.text,
},
...(isReplay && { _meta: { isReplay: true } }),
},
})
.catch((err) => {
Expand Down Expand Up @@ -774,11 +781,11 @@ export namespace ACP {
}

async setSessionModel(params: SetSessionModelRequest) {
const session = this.sessionManager.get(params.sessionId)
const session = await this.sessionManager.get(params.sessionId)

const model = Provider.parseModel(params.modelId)

this.sessionManager.setModel(session.id, {
await this.sessionManager.setModel(session.id, {
providerID: model.providerID,
modelID: model.modelID,
})
Expand All @@ -789,25 +796,25 @@ export namespace ACP {
}

async setSessionMode(params: SetSessionModeRequest): Promise<SetSessionModeResponse | void> {
this.sessionManager.get(params.sessionId)
await this.sessionManager.get(params.sessionId)
await this.config.sdk.app
.agents({}, { throwOnError: true })
.then((x) => x.data)
.then((agent) => {
if (!agent) throw new Error(`Agent not found: ${params.modeId}`)
})
this.sessionManager.setMode(params.sessionId, params.modeId)
await this.sessionManager.setMode(params.sessionId, params.modeId)
}

async prompt(params: PromptRequest) {
const sessionID = params.sessionId
const session = this.sessionManager.get(sessionID)
const session = await this.sessionManager.get(sessionID)
const directory = session.cwd

const current = session.model
const model = current ?? (await defaultModel(this.config, directory))
if (!current) {
this.sessionManager.setModel(session.id, model)
await this.sessionManager.setModel(session.id, model)
}
const agent = session.modeId ?? (await AgentModule.defaultAgent())

Expand Down Expand Up @@ -928,7 +935,7 @@ export namespace ACP {
}

async cancel(params: CancelNotification) {
const session = this.sessionManager.get(params.sessionId)
const session = await this.sessionManager.get(params.sessionId)
await this.config.sdk.session.abort(
{
sessionID: params.sessionId,
Expand Down
96 changes: 88 additions & 8 deletions packages/opencode/src/acp/session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,23 @@ import { RequestError, type McpServer } from "@agentclientprotocol/sdk"
import type { ACPSessionState } from "./types"
import { Log } from "@/util/log"
import type { OpencodeClient } from "@opencode-ai/sdk/v2"
import { Storage } from "@/storage/storage"

const log = Log.create({ service: "acp-session-manager" })

// Serializable version of ACPSessionState for disk persistence
interface PersistedACPSessionState {
id: string
cwd: string
mcpServers: McpServer[]
createdAt: string // ISO string instead of Date
model?: {
providerID: string
modelID: string
}
modeId?: string
}

export class ACPSessionManager {
private sessions = new Map<string, ACPSessionState>()
private sdk: OpencodeClient
Expand All @@ -13,6 +27,41 @@ export class ACPSessionManager {
this.sdk = sdk
}

/**
* Persist session state to disk for recovery after process restart
*/
private async persistState(state: ACPSessionState): Promise<void> {
const persisted: PersistedACPSessionState = {
...state,
createdAt: state.createdAt.toISOString(),
}
await Storage.write(["acp_state", state.id], persisted)
log.info("persisted_session_state", { sessionId: state.id })
}

/**
* Try to restore session state from disk
*/
private async restoreState(sessionId: string): Promise<ACPSessionState | undefined> {
try {
const persisted = await Storage.read<PersistedACPSessionState>(["acp_state", sessionId])
const state: ACPSessionState = {
...persisted,
createdAt: new Date(persisted.createdAt),
}
// Cache in memory after restore
this.sessions.set(sessionId, state)
log.info("restored_session_state", { sessionId })
return state
} catch (e) {
// Session not found on disk
if (e instanceof Storage.NotFoundError) {
return undefined
}
throw e
}
}

async create(cwd: string, mcpServers: McpServer[], model?: ACPSessionState["model"]): Promise<ACPSessionState> {
const session = await this.sdk.session
.create(
Expand All @@ -37,6 +86,7 @@ export class ACPSessionManager {
log.info("creating_session", { state })

this.sessions.set(sessionId, state)
await this.persistState(state)
return state
}

Expand Down Expand Up @@ -68,34 +118,64 @@ export class ACPSessionManager {
log.info("loading_session", { state })

this.sessions.set(sessionId, state)
await this.persistState(state)
return state
}

get(sessionId: string): ACPSessionState {
/**
* Get session from memory, or try to restore from disk if not found.
* This handles the case where the ACP process restarted but the session
* was previously created/loaded.
*/
async get(sessionId: string): Promise<ACPSessionState> {
// First, try in-memory cache
const cached = this.sessions.get(sessionId)
if (cached) {
return cached
}

// Try to restore from disk (handles process restart case)
const restored = await this.restoreState(sessionId)
if (restored) {
return restored
}

// Session truly doesn't exist
log.error("session not found", { sessionId })
throw RequestError.invalidParams(JSON.stringify({ error: `Session not found: ${sessionId}` }))
}

/**
* Synchronous get - only checks in-memory cache.
* Use this only when you're certain the session is already loaded.
*/
private getSync(sessionId: string): ACPSessionState {
const session = this.sessions.get(sessionId)
if (!session) {
log.error("session not found", { sessionId })
log.error("session not found in memory", { sessionId })
throw RequestError.invalidParams(JSON.stringify({ error: `Session not found: ${sessionId}` }))
}
return session
}

getModel(sessionId: string) {
const session = this.get(sessionId)
async getModel(sessionId: string) {
const session = await this.get(sessionId)
return session.model
}

setModel(sessionId: string, model: ACPSessionState["model"]) {
const session = this.get(sessionId)
async setModel(sessionId: string, model: ACPSessionState["model"]) {
const session = await this.get(sessionId)
session.model = model
this.sessions.set(sessionId, session)
await this.persistState(session)
return session
}

setMode(sessionId: string, modeId: string) {
const session = this.get(sessionId)
async setMode(sessionId: string, modeId: string) {
const session = await this.get(sessionId)
session.modeId = modeId
this.sessions.set(sessionId, session)
await this.persistState(session)
return session
}
}