agent-radar
2ed3415fb7
Publish Claude Code Binaries / Stage and publish binaries (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-arm64 (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build linux-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build windows-amd64 (push) Canceled after 0s
Sync Tutti Agent Skills / sync (push) Canceled after 0s
Windows Agent Adapters / Test Windows agent adapters (push) Canceled after 0s
Windows Daemon Adapters / Test Windows daemon adapters (push) Canceled after 0s
Publish Tutti App Runtime / Publish runtime catalog (push) Canceled after 0s
818 lines
29 KiB
Go
818 lines
29 KiB
Go
package agentruntime
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
|
|
"github.com/tutti-os/tutti/packages/agent/daemon/providerregistry"
|
|
)
|
|
|
|
// Codex may synchronously finish resource and thread initialization before
|
|
// replying to thread/start; keep this budget separate from generic ACP calls.
|
|
const codexAppServerThreadStartTimeout = 90 * time.Second
|
|
|
|
func (a *CodexAppServerAdapter) Start(ctx context.Context, session Session) (events []activityshared.Event, err error) {
|
|
unlockLifecycle := a.lockSessionLifecycle(session.AgentSessionID)
|
|
defer unlockLifecycle()
|
|
if err := a.admitCodexReplacementLocked(session.AgentSessionID); err != nil {
|
|
return nil, err
|
|
}
|
|
trace := newCodexAppServerStartupTrace(session, a.startupSpanObserver, a.startupObserver)
|
|
defer func() {
|
|
trace.Finish(err)
|
|
}()
|
|
extraSkillRoots, err := tuttiAgentExtraSkillRoots(a.config.skillRootsStrategy, session.Env)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stableSystemSkillsRoot, err := tuttiAgentStableSystemSkillsRoot(a.config.skillRootsStrategy, session.Env)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// One session owns at most one live app-server process. Starting over a
|
|
// session that already holds a live client replaces it: stop the old
|
|
// client first, then spawn the new process.
|
|
if existing := a.getSession(session.AgentSessionID); existing != nil && existing.client != nil {
|
|
a.rejectPendingRequests(session.AgentSessionID, errPermissionRequestCanceled)
|
|
_ = a.closeLiveSession(session.AgentSessionID)
|
|
}
|
|
client, initializeResult, _, err := a.startClient(ctx, session, trace, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
started := false
|
|
keepSession := false
|
|
startedSession := &codexAppServerSession{
|
|
client: client,
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
}
|
|
defer func() {
|
|
if !started {
|
|
a.closeOrRetainCodexSession(session.AgentSessionID, startedSession)
|
|
}
|
|
if !keepSession {
|
|
a.removeSession(session.AgentSessionID)
|
|
}
|
|
}()
|
|
serverInfo := a.appServerInfo(initializeResult)
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
client: client,
|
|
serverInfo: serverInfo,
|
|
acpLiveState: newACPLiveState(),
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
if err := a.stabilizeSystemSkillPaths(session, stableSystemSkillsRoot, trace); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := a.configureExtraSkillRoots(ctx, client, session, extraSkillRoots, trace); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
account, authRequired := a.fetchAccount(ctx, client, session, trace)
|
|
if authRequired {
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
serverInfo: serverInfo,
|
|
account: account,
|
|
authState: "auth_required",
|
|
authMessage: a.config.authRequiredMessage,
|
|
acpLiveState: newACPLiveState(),
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
keepSession = true
|
|
return []activityshared.Event{newSessionActivityEvent(session, EventSessionStarted, SessionStatusReady, map[string]any{
|
|
"adapter": a.commandString(),
|
|
"command": a.commandString(),
|
|
"agent": serverInfo,
|
|
"permissionModeId": session.PermissionModeID,
|
|
"authState": "auth_required",
|
|
"authMessage": a.config.authRequiredMessage,
|
|
})}, nil
|
|
}
|
|
models := []map[string]any(nil)
|
|
if codexAppServerNeedsSynchronousModels(session) {
|
|
models = a.fetchModels(ctx, client, session, trace)
|
|
}
|
|
if len(models) > 0 {
|
|
effectiveSettings := codexAppServerEffectiveSettings(models, session, nil)
|
|
session.Settings = &effectiveSettings
|
|
}
|
|
planModeMask, defaultModeMask := a.fetchCollaborationModeMasks(ctx, client, session, trace)
|
|
|
|
threadParams := appServerThreadStartParams(session, a.sessionCWD(session))
|
|
trace.Log("thread.start.params", codexAppServerTraceThreadStartParams(session, threadParams, false))
|
|
a.observeStartupResourcesAsync(session, client, trace)
|
|
callbackSession := session
|
|
threadResult, err := trace.TypedCall(codexAppServerThreadStartTimeout, appServerMethodThreadStart, func() (json.RawMessage, error) {
|
|
return client.ThreadStart(ctx, codexAppServerThreadStartTimeout, threadParams,
|
|
func(ctx context.Context, message acpMessage) error {
|
|
trace.LogMessage(message.Method, len(message.ID) > 0, len(message.Params))
|
|
_, err := a.handleAppServerMessage(ctx, client, callbackSession, "", message, nil, nil, nil)
|
|
return err
|
|
})
|
|
})
|
|
if err != nil {
|
|
var callErr *acpCallError
|
|
if errors.As(err, &callErr) && callErr.AuthRequired() {
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
serverInfo: serverInfo,
|
|
account: account,
|
|
authState: "auth_required",
|
|
authMessage: a.config.authRequiredMessage,
|
|
acpLiveState: newACPLiveState(),
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
keepSession = true
|
|
return []activityshared.Event{newSessionActivityEvent(session, EventSessionStarted, SessionStatusReady, map[string]any{
|
|
"adapter": a.commandString(),
|
|
"command": a.commandString(),
|
|
"agent": serverInfo,
|
|
"permissionModeId": session.PermissionModeID,
|
|
"authState": "auth_required",
|
|
"authMessage": a.config.authRequiredMessage,
|
|
})}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
threadID, err := appServerThreadID(threadResult)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
session.ProviderSessionID = threadID
|
|
trace.Log("thread.id.resolved", map[string]any{
|
|
"thread_id": threadID,
|
|
})
|
|
slog.Info("agent session app-server thread started",
|
|
"event", "agent_session.app_server.thread_start.succeeded",
|
|
"provider", a.config.provider,
|
|
"room_id", session.RoomID,
|
|
"agent_session_id", session.AgentSessionID,
|
|
"provider_session_id", threadID,
|
|
)
|
|
|
|
liveState := newACPLiveState()
|
|
liveState.currentMode = codexACPEffectiveModeID(session)
|
|
liveState.availableCommands = codexAppServerCommands()
|
|
liveState.commandsKnown = true
|
|
applyACPConfigOptionDescriptors(&liveState, codexAppServerConfigOptionDescriptors(models, session, threadResult))
|
|
|
|
started = true
|
|
keepSession = true
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
client: client,
|
|
threadID: threadID,
|
|
runtimeSession: session,
|
|
serverInfo: serverInfo,
|
|
account: account,
|
|
models: cloneCodexAppServerModels(models),
|
|
startupModelsReady: len(models) > 0,
|
|
startupRateLimitsReady: false,
|
|
planModeMask: planModeMask,
|
|
defaultModeMask: defaultModeMask,
|
|
defaultModel: codexAppServerSessionDefaultModel(session, models),
|
|
authState: "authenticated",
|
|
acpLiveState: liveState,
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
a.refreshStartupMetadataAsync(session, threadResult, len(models) == 0, a.config.rateLimits, trace)
|
|
a.emitCommandSnapshot(AgentSessionCommandSnapshot{
|
|
AgentSessionID: strings.TrimSpace(session.AgentSessionID),
|
|
Commands: codexAppServerCommands(),
|
|
})
|
|
return []activityshared.Event{newSessionActivityEvent(session, EventSessionStarted, SessionStatusReady, map[string]any{
|
|
"adapter": a.commandString(),
|
|
"command": a.commandString(),
|
|
"agent": serverInfo,
|
|
"permissionModeId": session.PermissionModeID,
|
|
})}, nil
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) Resume(ctx context.Context, session Session) (err error) {
|
|
if strings.TrimSpace(session.ProviderSessionID) == "" {
|
|
return missingProviderSessionResumeError(session)
|
|
}
|
|
unlockLifecycle := a.lockSessionLifecycle(session.AgentSessionID)
|
|
defer unlockLifecycle()
|
|
if err := a.admitCodexReplacementLocked(session.AgentSessionID); err != nil {
|
|
return err
|
|
}
|
|
// Resume may run over a session that still holds a live client. Unlike
|
|
// Start, the old client is kept alive until the replacement has resumed
|
|
// successfully (storeSession closes it on replace): if the new spawn or
|
|
// thread/resume fails, the previous session must remain usable.
|
|
trace := newCodexAppServerStartupTrace(session, a.startupSpanObserver, a.startupObserver)
|
|
defer func() {
|
|
trace.Finish(err)
|
|
}()
|
|
trace.Log("resume.begin", map[string]any{
|
|
"thread_id": strings.TrimSpace(session.ProviderSessionID),
|
|
})
|
|
extraSkillRoots, err := tuttiAgentExtraSkillRoots(a.config.skillRootsStrategy, session.Env)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
stableSystemSkillsRoot, err := tuttiAgentStableSystemSkillsRoot(a.config.skillRootsStrategy, session.Env)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
client, initializeResult, attachedCheckpoint, err := a.startClient(ctx, session, trace, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
started := false
|
|
keepSession := false
|
|
previousSession := a.getSession(session.AgentSessionID)
|
|
startedSession := &codexAppServerSession{
|
|
client: client,
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
}
|
|
defer func() {
|
|
if !started {
|
|
a.closeOrRetainCodexSession(session.AgentSessionID, startedSession)
|
|
}
|
|
if !keepSession {
|
|
if previousSession != nil {
|
|
a.storeSession(session.AgentSessionID, previousSession)
|
|
} else {
|
|
a.removeSession(session.AgentSessionID)
|
|
}
|
|
}
|
|
}()
|
|
if attachedCheckpoint {
|
|
planModeMask, defaultModeMask, defaultModel, checkpointFound :=
|
|
codexAppServerProtocolCheckpointFromRuntimeContext(session.RuntimeContext)
|
|
if !checkpointFound {
|
|
return errors.New(
|
|
"attached live Codex replay is missing provider resume checkpoint; record a new cassette",
|
|
)
|
|
}
|
|
liveState := newACPLiveState()
|
|
liveState.currentMode = codexACPEffectiveModeID(session)
|
|
liveState.availableCommands = codexAppServerCommands()
|
|
liveState.commandsKnown = true
|
|
applyACPConfigOptionDescriptors(
|
|
&liveState,
|
|
codexAppServerConfigOptionDescriptors(nil, session, nil),
|
|
)
|
|
started = true
|
|
keepSession = true
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
client: client,
|
|
threadID: strings.TrimSpace(session.ProviderSessionID),
|
|
resumeRuntimeContext: clonePayload(session.RuntimeContext),
|
|
planModeMask: planModeMask,
|
|
defaultModeMask: defaultModeMask,
|
|
defaultModel: firstNonEmpty(
|
|
strings.TrimSpace(session.SettingsValue().Model),
|
|
defaultModel,
|
|
),
|
|
authState: "authenticated",
|
|
acpLiveState: liveState,
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
a.emitCommandSnapshot(AgentSessionCommandSnapshot{
|
|
AgentSessionID: strings.TrimSpace(session.AgentSessionID),
|
|
Commands: codexAppServerCommands(),
|
|
})
|
|
return nil
|
|
}
|
|
serverInfo := a.appServerInfo(initializeResult)
|
|
if err := a.stabilizeSystemSkillPaths(session, stableSystemSkillsRoot, trace); err != nil {
|
|
return err
|
|
}
|
|
if err := a.configureExtraSkillRoots(ctx, client, session, extraSkillRoots, trace); err != nil {
|
|
return err
|
|
}
|
|
|
|
account, authRequired := a.fetchAccount(ctx, client, session, trace)
|
|
if authRequired {
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
threadID: session.ProviderSessionID,
|
|
serverInfo: serverInfo,
|
|
account: account,
|
|
authState: "auth_required",
|
|
authMessage: a.config.authRequiredMessage,
|
|
acpLiveState: newACPLiveState(),
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
keepSession = true
|
|
return nil
|
|
}
|
|
models := []map[string]any(nil)
|
|
if codexAppServerNeedsSynchronousModels(session) {
|
|
models = a.fetchModels(ctx, client, session, trace)
|
|
}
|
|
if len(models) > 0 && strings.TrimSpace(session.SettingsValue().ReasoningEffort) != "" {
|
|
hasExplicitModel := strings.TrimSpace(session.SettingsValue().Model) != ""
|
|
effectiveSettings := codexAppServerEffectiveSettings(models, session, nil)
|
|
// The catalog default is needed to validate an effort-only persisted
|
|
// setting, but it must not become a thread/resume model override. The
|
|
// existing thread remains authoritative until the resume result reports
|
|
// its actual model.
|
|
if !hasExplicitModel {
|
|
effectiveSettings.Model = ""
|
|
}
|
|
session.Settings = &effectiveSettings
|
|
}
|
|
planModeMask, defaultModeMask := a.fetchCollaborationModeMasks(ctx, client, session, trace)
|
|
|
|
params := appServerThreadStartParams(session, a.sessionCWD(session))
|
|
params["threadId"] = strings.TrimSpace(session.ProviderSessionID)
|
|
trace.Log("thread.start.params", codexAppServerTraceThreadStartParams(session, params, true))
|
|
a.observeStartupResourcesAsync(session, client, trace)
|
|
// codex replays thread/tokenUsage/updated during thread/resume so the GUI
|
|
// can show context fill before a new turn runs. The resumed session is not
|
|
// stored yet, so applyTokenUsage cannot reach it; capture the replayed
|
|
// usage here and fold it into the live state below.
|
|
var replayedUsage acpUsageState
|
|
replayedUsageKnown := false
|
|
callbackSession := session
|
|
threadResult, err := trace.TypedCall(acpStartCallTimeout, appServerMethodThreadResume, func() (json.RawMessage, error) {
|
|
return client.ThreadResume(ctx, acpStartCallTimeout, params,
|
|
func(ctx context.Context, message acpMessage) error {
|
|
trace.LogMessage(message.Method, len(message.ID) > 0, len(message.Params))
|
|
if message.Method == appServerNotifyTokenUsage && len(message.Params) > 0 {
|
|
tokenParams := map[string]any{}
|
|
if json.Unmarshal(message.Params, &tokenParams) == nil {
|
|
if usage, ok := appServerTokenUsageState(tokenParams); ok {
|
|
replayedUsage = usage
|
|
replayedUsageKnown = true
|
|
}
|
|
}
|
|
}
|
|
_, err := a.handleAppServerMessage(ctx, client, callbackSession, "", message, nil, nil, nil)
|
|
return err
|
|
})
|
|
})
|
|
if err != nil {
|
|
return classifyACPResumeError(session, appServerMethodThreadResume, err)
|
|
}
|
|
if len(models) > 0 {
|
|
effectiveSettings := codexAppServerEffectiveSettings(models, session, threadResult)
|
|
session.Settings = &effectiveSettings
|
|
}
|
|
liveState := newACPLiveState()
|
|
liveState.currentMode = codexACPEffectiveModeID(session)
|
|
liveState.availableCommands = codexAppServerCommands()
|
|
liveState.commandsKnown = true
|
|
applyACPConfigOptionDescriptors(&liveState, codexAppServerConfigOptionDescriptors(models, session, threadResult))
|
|
if replayedUsageKnown {
|
|
liveState.usage = mergeACPUsageState(liveState.usage, replayedUsage)
|
|
}
|
|
|
|
started = true
|
|
keepSession = true
|
|
a.storeSession(session.AgentSessionID, &codexAppServerSession{
|
|
client: client,
|
|
threadID: strings.TrimSpace(session.ProviderSessionID),
|
|
runtimeSession: session,
|
|
serverInfo: serverInfo,
|
|
account: account,
|
|
models: cloneCodexAppServerModels(models),
|
|
startupModelsReady: len(models) > 0,
|
|
startupRateLimitsReady: false,
|
|
planModeMask: planModeMask,
|
|
defaultModeMask: defaultModeMask,
|
|
defaultModel: codexAppServerSessionDefaultModel(session, models),
|
|
authState: "authenticated",
|
|
acpLiveState: liveState,
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
a.refreshStartupMetadataAsync(session, threadResult, len(models) == 0, a.config.rateLimits, trace)
|
|
// Mirror Start: push the command snapshot so a resumed session advertises
|
|
// review/compact/undo to the GUI (otherwise the slash palette and the
|
|
// review picker only work on freshly created sessions).
|
|
a.emitCommandSnapshot(AgentSessionCommandSnapshot{
|
|
AgentSessionID: strings.TrimSpace(session.AgentSessionID),
|
|
Commands: codexAppServerCommands(),
|
|
})
|
|
return nil
|
|
}
|
|
|
|
func (*CodexAppServerAdapter) CanResume(session Session) bool {
|
|
return strings.TrimSpace(session.ProviderSessionID) != ""
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) HasLiveSession(session Session) bool {
|
|
a.mu.Lock()
|
|
appSession := a.sessions[strings.TrimSpace(session.AgentSessionID)]
|
|
if appSession == nil || appSession.client == nil || appSession.releasing || appSession.releaseFailed {
|
|
a.mu.Unlock()
|
|
return false
|
|
}
|
|
client := appSession.client
|
|
a.mu.Unlock()
|
|
select {
|
|
case <-client.Done():
|
|
return false
|
|
default:
|
|
return true
|
|
}
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) Close(ctx context.Context, session Session) error {
|
|
if a == nil {
|
|
return nil
|
|
}
|
|
agentSessionID := strings.TrimSpace(session.AgentSessionID)
|
|
unlockLifecycle := a.lockSessionLifecycle(agentSessionID)
|
|
defer unlockLifecycle()
|
|
a.rejectPendingRequests(agentSessionID, errPermissionRequestCanceled)
|
|
if appSession := a.getSession(agentSessionID); appSession != nil &&
|
|
appSession.client != nil &&
|
|
strings.TrimSpace(appSession.threadID) != "" {
|
|
a.mu.Lock()
|
|
shared := a.clientReferencedLocked(appSession.client, agentSessionID)
|
|
a.mu.Unlock()
|
|
if session.IsSideConversation() || shared {
|
|
if err := appSession.client.ThreadUnsubscribeNoHandler(
|
|
ctx,
|
|
acpStartCallTimeout,
|
|
appSession.threadID,
|
|
); err != nil {
|
|
select {
|
|
case <-appSession.client.Done():
|
|
// The shared process is already gone; local removal is the
|
|
// only cleanup left and Side will be expired.
|
|
default:
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return a.closeLiveSession(agentSessionID)
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) QuiesceForClose(
|
|
ctx context.Context,
|
|
session Session,
|
|
) error {
|
|
if a == nil {
|
|
return nil
|
|
}
|
|
appTurn := a.sessionActiveTurn(session.AgentSessionID)
|
|
if appTurn == nil &&
|
|
a.sessionActiveTurnID(session.AgentSessionID) == "" {
|
|
return nil
|
|
}
|
|
appSession := a.getSession(session.AgentSessionID)
|
|
_, err := a.Cancel(ctx, session, "session closed")
|
|
if errors.Is(err, ErrSessionDisconnected) {
|
|
return nil
|
|
}
|
|
if err != nil && !errors.Is(err, ErrSessionNoActiveTurn) {
|
|
return err
|
|
}
|
|
if appTurn == nil {
|
|
return nil
|
|
}
|
|
select {
|
|
case <-appTurn.terminated:
|
|
return nil
|
|
default:
|
|
}
|
|
|
|
// Cancel queues an interrupt when turn/start has been sent but has not
|
|
// returned the provider Turn id yet. Close must not detach the session while
|
|
// that queued interrupt still depends on the session registry. Wait for the
|
|
// normal binding/interrupt path, then tear down the shared transport if the
|
|
// provider never supplies an interruptible identity within the ordinary
|
|
// cancellation grace window.
|
|
grace := a.cancelGraceWindow
|
|
if grace <= 0 {
|
|
grace = defaultCodexAppServerCancelGraceWindow
|
|
}
|
|
timer := time.NewTimer(grace)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-appTurn.terminated:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-timer.C:
|
|
}
|
|
a.markTurnForceCanceled(appTurn)
|
|
slog.Warn(
|
|
"agent session app-server force-closing turn with unresolved provider identity",
|
|
"event", "agent_session.app_server.close.pending_turn_start_forced",
|
|
"agent_session_id", session.AgentSessionID,
|
|
"provider_session_id", session.ProviderSessionID,
|
|
"turn_id", appTurn.turnID,
|
|
"grace_ms", grace.Milliseconds(),
|
|
)
|
|
if appSession != nil && appSession.client != nil {
|
|
_ = appSession.client.Close()
|
|
}
|
|
select {
|
|
case <-appTurn.terminated:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) ReleaseLiveSession(_ context.Context, session Session) error {
|
|
if a == nil {
|
|
return nil
|
|
}
|
|
agentSessionID := strings.TrimSpace(session.AgentSessionID)
|
|
unlockLifecycle := a.lockSessionLifecycle(agentSessionID)
|
|
defer unlockLifecycle()
|
|
if a.hasLiveSessionWork(agentSessionID) {
|
|
return ErrLiveSessionBusy
|
|
}
|
|
return a.closeLiveSession(agentSessionID)
|
|
}
|
|
|
|
// DisconnectLiveSession resolves pending interactions and drops only the
|
|
// app-server transport. The Codex thread remains resumable; no provider
|
|
// thread/session deletion request is sent.
|
|
func (a *CodexAppServerAdapter) DisconnectLiveSession(_ context.Context, session Session) error {
|
|
if a == nil {
|
|
return nil
|
|
}
|
|
agentSessionID := strings.TrimSpace(session.AgentSessionID)
|
|
unlockLifecycle := a.lockSessionLifecycle(agentSessionID)
|
|
defer unlockLifecycle()
|
|
a.rejectPendingRequests(agentSessionID, ErrSessionDisconnected)
|
|
return a.closeLiveSession(agentSessionID)
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) closeLiveSession(agentSessionID string) error {
|
|
a.mu.Lock()
|
|
appSession := a.sessions[agentSessionID]
|
|
shared := appSession != nil &&
|
|
a.clientReferencedLocked(appSession.client, agentSessionID)
|
|
if shared {
|
|
delete(a.sessions, agentSessionID)
|
|
a.mu.Unlock()
|
|
return nil
|
|
}
|
|
if appSession != nil && appSession.client != nil {
|
|
appSession.releasing = true
|
|
appSession.client.SetMessageHandler(nil)
|
|
}
|
|
a.mu.Unlock()
|
|
if appSession != nil && appSession.client != nil {
|
|
if err := appSession.client.Close(); err != nil {
|
|
if codexSessionAlreadyGone(err) {
|
|
a.mu.Lock()
|
|
if a.sessions[agentSessionID] == appSession {
|
|
delete(a.sessions, agentSessionID)
|
|
}
|
|
a.mu.Unlock()
|
|
return nil
|
|
}
|
|
a.mu.Lock()
|
|
if a.sessions[agentSessionID] == appSession {
|
|
appSession.releasing = false
|
|
appSession.releaseFailed = true
|
|
}
|
|
a.mu.Unlock()
|
|
return err
|
|
}
|
|
a.mu.Lock()
|
|
if a.sessions[agentSessionID] == appSession {
|
|
delete(a.sessions, agentSessionID)
|
|
}
|
|
a.mu.Unlock()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) startInitializedClient(
|
|
ctx context.Context,
|
|
session Session,
|
|
trace *codexAppServerStartupTrace,
|
|
) (*codexAppServerClient, json.RawMessage, error) {
|
|
client, initializeResult, _, err := a.startClient(ctx, session, trace, false)
|
|
return client, initializeResult, err
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) startClient(
|
|
ctx context.Context,
|
|
session Session,
|
|
trace *codexAppServerStartupTrace,
|
|
allowAttachedCheckpoint bool,
|
|
) (*codexAppServerClient, json.RawMessage, bool, error) {
|
|
spec, cleanup, err := a.prepareInitializedClientLaunch(ctx, session)
|
|
if err != nil {
|
|
trace.Log("process.prepare.failed", map[string]any{
|
|
"error": err.Error(),
|
|
})
|
|
return nil, nil, false, err
|
|
}
|
|
return a.startClientPrepared(
|
|
ctx,
|
|
session,
|
|
trace,
|
|
spec,
|
|
cleanup,
|
|
allowAttachedCheckpoint,
|
|
)
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) prepareInitializedClientLaunch(
|
|
ctx context.Context,
|
|
session Session,
|
|
) (ProcessSpec, func(context.Context), error) {
|
|
if a == nil || a.transport == nil {
|
|
return ProcessSpec{}, nil, errors.New(
|
|
"app-server process transport is unavailable",
|
|
)
|
|
}
|
|
command := append([]string(nil), a.config.command...)
|
|
spawnEnv := append(codexACPEnv(session, a.host), session.Env...)
|
|
if a.commandResolver != nil {
|
|
resolved, err := a.commandResolver(ctx, a.config.provider)
|
|
if err != nil {
|
|
return ProcessSpec{}, nil, err
|
|
}
|
|
if len(resolved.Command) > 0 {
|
|
command = append([]string(nil), resolved.Command...)
|
|
}
|
|
spawnEnv = append(spawnEnv, resolved.Env...)
|
|
}
|
|
spec, cleanup, err := prepareProviderLaunch(ctx, a.preparer, session, ProcessSpec{
|
|
Provider: a.config.provider,
|
|
AgentSessionID: session.AgentSessionID,
|
|
RootAgentSessionID: session.RootAgentSessionID,
|
|
RoomID: session.RoomID,
|
|
CWD: a.sessionCWD(session),
|
|
Command: command,
|
|
Env: spawnEnv,
|
|
})
|
|
if err != nil {
|
|
return ProcessSpec{}, nil, err
|
|
}
|
|
if a.config.skillRootsStrategy == providerregistry.AppServerSkillRootsStrategyTuttiStable {
|
|
spec.Env = withoutEnvironmentKey(spec.Env, tuttiAgentExtraSkillRootsEnv)
|
|
spec.Env = withoutEnvironmentKey(spec.Env, tuttiAgentStableSystemSkillsEnv)
|
|
}
|
|
spec.Env = withCodexAppServerLogging(spec.Env)
|
|
return spec, cleanup, nil
|
|
}
|
|
|
|
func (a *CodexAppServerAdapter) startClientPrepared(
|
|
ctx context.Context,
|
|
session Session,
|
|
trace *codexAppServerStartupTrace,
|
|
spec ProcessSpec,
|
|
cleanup func(context.Context),
|
|
allowAttachedCheckpoint bool,
|
|
) (*codexAppServerClient, json.RawMessage, bool, error) {
|
|
trace.Log("process.start.begin", map[string]any{
|
|
"command": strings.Join(spec.Command, " "),
|
|
"cwd": spec.CWD,
|
|
})
|
|
processStartedAt := time.Now()
|
|
conn, err := a.transport.Start(ctx, spec)
|
|
if err != nil {
|
|
cleanupPreparedLaunch(cleanup)
|
|
trace.Log("process.start.failed", map[string]any{
|
|
"duration_ms": time.Since(processStartedAt).Milliseconds(),
|
|
"error": err.Error(),
|
|
})
|
|
return nil, nil, false, err
|
|
}
|
|
conn = wrapProviderLaunchCleanup(conn, cleanup)
|
|
trace.Log("process.start.succeeded", map[string]any{
|
|
"duration_ms": time.Since(processStartedAt).Milliseconds(),
|
|
})
|
|
client := newCodexAppServerClient(conn)
|
|
client.SetStderrSink(trace.LogStderr)
|
|
// The session-level handler receives every message that arrives outside
|
|
// an in-flight RPC. Because turn/start responds immediately while the
|
|
// turn keeps streaming, this is the main delivery path for turn output:
|
|
// resolve the active turn context so notifications keep producing
|
|
// activity events after the RPC has returned. Child sessions may outlive
|
|
// that root turn, so their events retain a session-level fallback below.
|
|
client.SetMessageHandler(func(ctx context.Context, message acpMessage) error {
|
|
endInputUnit := a.inputUnits.begin(ctx, session.AgentSessionID)
|
|
defer endInputUnit()
|
|
turnSession := session
|
|
if appSession := a.getSession(session.AgentSessionID); appSession != nil {
|
|
turnSession.ProviderSessionID = firstNonEmpty(appSession.threadID, turnSession.ProviderSessionID)
|
|
}
|
|
turnID := ""
|
|
var normalizer *acpTurnNormalizer
|
|
var turnEmit func([]activityshared.Event)
|
|
var turnEmitCommands CommandSnapshotSink
|
|
if activeTurn := a.sessionActiveTurn(session.AgentSessionID); activeTurn != nil {
|
|
turnSession = activeTurn.session
|
|
turnID = activeTurn.turnID
|
|
normalizer = activeTurn.normalizer
|
|
turnEmit = activeTurn.emit
|
|
turnEmitCommands = activeTurn.emitCommands
|
|
}
|
|
events, err := a.handleAppServerMessage(ctx, client, turnSession, turnID, message, normalizer, turnEmit, turnEmitCommands)
|
|
// Stamp the reduction while the decoder-owned provider input unit is
|
|
// still in scope. Most turn output is stamped again by the active-turn
|
|
// emitter, but lifecycle reductions such as turn/started can take a
|
|
// different emission path. The tracker is idempotent, so the emitter
|
|
// can retain its existing boundary without duplicating indexes.
|
|
events = a.inputUnits.stamp(session.AgentSessionID, events)
|
|
// A child may outlive its root Turn. Only child events owned by the
|
|
// currently active canonical root Turn may enter that Turn's emitter;
|
|
// otherwise a newer Turn's acceptance buffer or close fence can drop them.
|
|
turnEvents, detachedChildEvents := appServerEventsForActiveRootTurn(
|
|
session.AgentSessionID,
|
|
turnID,
|
|
events,
|
|
)
|
|
if turnEmit != nil {
|
|
turnEmit(turnEvents)
|
|
}
|
|
if len(detachedChildEvents) > 0 {
|
|
a.emitSessionEvents(
|
|
session.AgentSessionID,
|
|
a.stampTurnLifecycleSnapshots(session.AgentSessionID, detachedChildEvents),
|
|
)
|
|
}
|
|
return err
|
|
})
|
|
started := false
|
|
defer func() {
|
|
if !started {
|
|
a.closeOrRetainCodexSession(session.AgentSessionID, &codexAppServerSession{
|
|
client: client,
|
|
pendingRequests: make(map[string]*pendingInteractiveRequest),
|
|
})
|
|
}
|
|
}()
|
|
captureOrigin := processCassetteCaptureOrigin(conn)
|
|
if captureOrigin == ProcessCassetteCaptureOriginAttachedLiveConnection {
|
|
if !allowAttachedCheckpoint {
|
|
return nil, nil, false, errors.New(
|
|
"attached live provider checkpoint cannot start a new Codex session",
|
|
)
|
|
}
|
|
started = true
|
|
return client, nil, true, nil
|
|
}
|
|
|
|
initializeResult, err := trace.TypedCall(acpStartCallTimeout, appServerMethodInitialize, func() (json.RawMessage, error) {
|
|
return client.Initialize(ctx, acpStartCallTimeout, map[string]any{
|
|
"clientInfo": a.clientInfoParams(spec.Env),
|
|
"capabilities": map[string]any{
|
|
"experimentalApi": true,
|
|
},
|
|
}, func(ctx context.Context, message acpMessage) error {
|
|
trace.LogMessage(message.Method, len(message.ID) > 0, len(message.Params))
|
|
_, err := a.handleAppServerMessage(ctx, client, session, "", message, nil, nil, nil)
|
|
return err
|
|
})
|
|
})
|
|
if err != nil {
|
|
slog.Warn("agent session app-server initialize failed",
|
|
"event", "agent_session.app_server.initialize.failed",
|
|
"room_id", session.RoomID,
|
|
"agent_session_id", session.AgentSessionID,
|
|
"error", err.Error(),
|
|
)
|
|
return nil, nil, false, err
|
|
}
|
|
trace.Log("initialized.notify.begin", nil)
|
|
notifyStartedAt := time.Now()
|
|
if err := client.Initialized(ctx); err != nil {
|
|
trace.Log("initialized.notify.failed", map[string]any{
|
|
"duration_ms": time.Since(notifyStartedAt).Milliseconds(),
|
|
"error": err.Error(),
|
|
})
|
|
return nil, nil, false, err
|
|
}
|
|
trace.Log("initialized.notify.succeeded", map[string]any{
|
|
"duration_ms": time.Since(notifyStartedAt).Milliseconds(),
|
|
})
|
|
started = true
|
|
return client, initializeResult, false, nil
|
|
}
|
|
|
|
func appServerEventsForActiveRootTurn(
|
|
rootAgentSessionID string,
|
|
activeRootTurnID string,
|
|
events []activityshared.Event,
|
|
) ([]activityshared.Event, []activityshared.Event) {
|
|
rootAgentSessionID = strings.TrimSpace(rootAgentSessionID)
|
|
activeRootTurnID = strings.TrimSpace(activeRootTurnID)
|
|
turnEvents := make([]activityshared.Event, 0, len(events))
|
|
detachedChildEvents := make([]activityshared.Event, 0, len(events))
|
|
for _, event := range events {
|
|
eventAgentSessionID := strings.TrimSpace(event.AgentSessionID)
|
|
if eventAgentSessionID != "" && eventAgentSessionID != rootAgentSessionID {
|
|
if activeRootTurnID == "" || strings.TrimSpace(event.RootTurnID) != activeRootTurnID {
|
|
detachedChildEvents = append(detachedChildEvents, event)
|
|
continue
|
|
}
|
|
}
|
|
if activeRootTurnID != "" {
|
|
turnEvents = append(turnEvents, event)
|
|
}
|
|
}
|
|
return turnEvents, detachedChildEvents
|
|
}
|