package agenthost import ( "context" "log/slog" "strings" storesqlite "github.com/tutti-os/tutti/packages/agent/store-sqlite" canonical "github.com/tutti-os/tutti/packages/agent/store-sqlite/canonical" ) type ActivityStateCommitted struct { Input canonical.ReportSessionStateInput Reply canonical.ReportSessionStateReply Result storesqlite.ActivityStateReportResult } type SessionMessagesCommitted struct { Input canonical.ReportSessionMessagesInput Reply canonical.ReportSessionMessagesReply Result storesqlite.MessageReportResult // Provider is the canonical provider identity for the reporting session. // Message rows do not repeat session identity, so terminal tool-call // analytics carry it on the committed wrapper. Provider string // IsChildSession marks a provider-native subagent session. Message reports // carry no session state, so the reporting adapter resolves it from the // canonical session before publishing the delta. IsChildSession bool } type RootTurnSettled struct { WorkspaceID string AgentSessionID string Turn storesqlite.Turn Provider string IsChildSession bool // StartupReconciled marks a turn force-settled by daemon-start // reconciliation rather than by a live provider settlement. Failure // analytics still count it; runtime fan-out that reacts to a real // completion must skip it. StartupReconciled bool } type RuntimeOperationCommitStage string const ( RuntimeOperationPrepared RuntimeOperationCommitStage = "prepared" RuntimeOperationCheckpoint RuntimeOperationCommitStage = "checkpointed" RuntimeOperationCompleted RuntimeOperationCommitStage = "completed" RuntimeOperationReleased RuntimeOperationCommitStage = "released" RuntimeOperationFailed RuntimeOperationCommitStage = "failed" ) type RuntimeOperationCommitted struct { Stage RuntimeOperationCommitStage Operation storesqlite.RuntimeOperation Event *storesqlite.RuntimeOperationEvent Provider string IsChildSession bool } type GoalOperationCommitStage string const ( GoalOperationPrepared GoalOperationCommitStage = "prepared" GoalOperationDispatched GoalOperationCommitStage = "dispatched" GoalOperationAcknowledged GoalOperationCommitStage = "acknowledged" GoalOperationCompleted GoalOperationCommitStage = "completed" GoalOperationReleased GoalOperationCommitStage = "released" GoalOperationFailed GoalOperationCommitStage = "failed" GoalOperationEvidence GoalOperationCommitStage = "evidence" GoalOperationReconciled GoalOperationCommitStage = "reconciled" GoalOperationRepairPrepared GoalOperationCommitStage = "repair_prepared" GoalOperationTerminal GoalOperationCommitStage = "terminal" ) type GoalOperationCommitted struct { Stage GoalOperationCommitStage Operation storesqlite.GoalControlOperation State storesqlite.SessionGoalState Audit *storesqlite.Message Provider string IsChildSession bool } type CanonicalProjectionDirty struct { WorkspaceID string AgentSessionID string RootAgentSessionID string RootTurnID string MutationID string EntityKind string EntityID string Operation string Version int64 } type CanonicalViewInvalidated struct { WorkspaceID string AgentSessionID string } // CommittedDelta describes facts that have already committed. ProjectionDirty // is a wake hint for a durable outbox marker written by a transaction // participant; it is never the durable marker itself. type CommittedDelta struct { TransactionID string ActivityState *ActivityStateCommitted SessionMessages *SessionMessagesCommitted RootTurnsSettled []RootTurnSettled RuntimeOperation *RuntimeOperationCommitted GoalOperation *GoalOperationCommitted ProjectionDirty []CanonicalProjectionDirty ViewsInvalidated []CanonicalViewInvalidated } func ActivityStateDelta(input canonical.ReportSessionStateInput, reply canonical.ReportSessionStateReply, result storesqlite.ActivityStateReportResult) CommittedDelta { delta := committedDeltaFromMutations(result.TransactionID, result.CommitDelta.Mutations) delta.ActivityState = &ActivityStateCommitted{Input: input, Reply: reply, Result: result} if result.RootTurnAccepted && result.RootTurn.Phase == storesqlite.TurnPhaseSettled { delta.RootTurnsSettled = append(delta.RootTurnsSettled, RootTurnSettled{ WorkspaceID: result.RootTurn.WorkspaceID, AgentSessionID: result.RootTurn.AgentSessionID, Turn: result.RootTurn, Provider: firstNonEmptyTrimmed(result.State.Session.Provider, input.State.Provider, input.Source.Provider), IsChildSession: rootTurnOwnerIsChild(input, result), }) } if goalOp := goalOperationFromActivityState(input, result); goalOp != nil { delta.GoalOperation = goalOp } delta.addView(input.WorkspaceID, input.AgentSessionID) if result.RootTurnAccepted { delta.addView(input.WorkspaceID, result.RootTurn.AgentSessionID) } return delta } func rootTurnOwnerIsChild( input canonical.ReportSessionStateInput, result storesqlite.ActivityStateReportResult, ) bool { ownerSessionID := strings.TrimSpace(result.RootTurn.AgentSessionID) reportingSessionID := strings.TrimSpace(result.State.Session.ID) if reportingSessionID == "" { reportingSessionID = strings.TrimSpace(input.AgentSessionID) } if ownerSessionID == "" || ownerSessionID != reportingSessionID { // A child terminal report may settle its canonical root Turn. The // persisted Session in this result still belongs to the reporting child, // so its kind must not be projected onto the root owner. return false } if strings.TrimSpace(result.State.Session.ID) != "" { return canonicalSessionIsChild(result.State.Session) } return sessionStateIsChild(input.State) } func sessionStateIsChild(state canonical.WorkspaceAgentSessionStateUpdate) bool { kind := strings.TrimSpace(state.Kind) if strings.EqualFold(kind, storesqlite.SessionKindChild) { return true } return strings.TrimSpace(state.ParentToolCallID) != "" } // goalOperationFromActivityState promotes bottom-up Goal observed updates // (reconcileObservedGoalFromSessionTx) into the typed GoalOperation seam so // session-replay can mint goal.completed / goal.running checkpoints. func goalOperationFromActivityState( input canonical.ReportSessionStateInput, result storesqlite.ActivityStateReportResult, ) *GoalOperationCommitted { sessionID := strings.TrimSpace(input.AgentSessionID) workspaceID := strings.TrimSpace(input.WorkspaceID) hasGoalMutation := false for _, mutation := range result.CommitDelta.Mutations { if mutation.EntityKind != storesqlite.MutationEntityGoalState { continue } hasGoalMutation = true if id := strings.TrimSpace(mutation.AgentSessionID); id != "" { sessionID = id } if id := strings.TrimSpace(mutation.WorkspaceID); id != "" { workspaceID = id } } if !hasGoalMutation || sessionID == "" || workspaceID == "" { return nil } observed := clonePayload(runtimeContextGoal(input.State.RuntimeContext)) if len(observed) == 0 { return nil } return &GoalOperationCommitted{ Stage: GoalOperationReconciled, Provider: firstNonEmptyTrimmed(result.State.Session.Provider, input.State.Provider, input.Source.Provider), IsChildSession: sessionStateIsChild(input.State), State: storesqlite.SessionGoalState{ WorkspaceID: workspaceID, AgentSessionID: sessionID, Observed: observed, CommitTransactionID: strings.TrimSpace(result.TransactionID), }, } } func runtimeContextGoal(runtimeContext map[string]any) map[string]any { if len(runtimeContext) == 0 { return nil } raw, ok := runtimeContext["goal"].(map[string]any) if !ok || len(raw) == 0 { return nil } return raw } func SessionMessagesDelta(input canonical.ReportSessionMessagesInput, reply canonical.ReportSessionMessagesReply, result storesqlite.MessageReportResult) CommittedDelta { delta := committedDeltaFromMutations(result.TransactionID, result.CommitDelta.Mutations) delta.SessionMessages = &SessionMessagesCommitted{ Input: input, Reply: reply, Result: result, Provider: strings.TrimSpace(input.Source.Provider), } delta.addView(input.WorkspaceID, canonicalMessageSessionID(input.AgentSessionID, result.Messages)) return delta } // StaleTurnSettlementDelta projects daemon-start reconciliation. Every listed // turn was force-settled as interrupted, so it enters RootTurnsSettled with // that outcome and the StartupReconciled marker. func StaleTurnSettlementDelta(settlements []storesqlite.StaleTurnSettlement) CommittedDelta { delta := CommittedDelta{} if len(settlements) > 0 { delta = committedDeltaFromMutations(settlements[0].TransactionID, settlements[0].CommitDelta.Mutations) } for _, settlement := range settlements { delta.addView(settlement.WorkspaceID, settlement.AgentSessionID) turn := settlement.Turn turnID := firstNonEmptyTrimmed(turn.TurnID, settlement.TurnID) if turnID == "" { continue } workspaceID := firstNonEmptyTrimmed(turn.WorkspaceID, settlement.WorkspaceID) agentSessionID := firstNonEmptyTrimmed(turn.AgentSessionID, settlement.AgentSessionID) if strings.TrimSpace(turn.TurnID) == "" { // Compatibility for callers that construct the legacy scalar-only // settlement value. The SQLite startup path always supplies Turn. turn = storesqlite.Turn{ WorkspaceID: workspaceID, AgentSessionID: agentSessionID, TurnID: turnID, Phase: storesqlite.TurnPhaseSettled, Outcome: storesqlite.TurnOutcomeInterrupted, ErrorMessage: "stale turn settled on daemon startup", StartedAtUnixMS: settlement.StartedAtUnixMS, SettledAtUnixMS: settlement.SettledAtUnixMS, } } delta.RootTurnsSettled = append(delta.RootTurnsSettled, RootTurnSettled{ WorkspaceID: workspaceID, AgentSessionID: agentSessionID, Turn: turn, Provider: strings.TrimSpace(settlement.Provider), IsChildSession: settlement.IsChildSession, StartupReconciled: true, }) } return delta } // CanonicalDelta exposes a post-commit canonical mutation without inventing a // command-specific Host event. Adapters use its projection-dirty identities // and view invalidations as wake hints only. func CanonicalDelta(commit storesqlite.TransactionDelta) CommittedDelta { return committedDeltaFromMutations(commit.TransactionID, commit.Mutations) } func runtimeOperationDelta(stage RuntimeOperationCommitStage, operation storesqlite.RuntimeOperation, event *storesqlite.RuntimeOperationEvent) CommittedDelta { delta := committedDeltaFromMutations(operation.CommitTransactionID, operation.CommitDelta.Mutations) delta.RuntimeOperation = &RuntimeOperationCommitted{Stage: stage, Operation: operation, Event: event} return delta } func goalOperationDelta(stage GoalOperationCommitStage, operation storesqlite.GoalControlOperation, state storesqlite.SessionGoalState, audit *storesqlite.Message) CommittedDelta { transactionID := operation.CommitTransactionID mutations := operation.CommitDelta.Mutations if transactionID == "" { transactionID = state.CommitTransactionID mutations = state.CommitDelta.Mutations } delta := committedDeltaFromMutations(transactionID, mutations) delta.GoalOperation = &GoalOperationCommitted{Stage: stage, Operation: operation, State: state, Audit: audit} delta.addView(operation.WorkspaceID, operation.AgentSessionID) if operation.AgentSessionID == "" { delta.addView(state.WorkspaceID, state.AgentSessionID) } return delta } func committedDeltaFromMutations(transactionID string, mutations []storesqlite.TransactionMutation) CommittedDelta { delta := CommittedDelta{TransactionID: strings.TrimSpace(transactionID)} for _, mutation := range mutations { delta.ProjectionDirty = append(delta.ProjectionDirty, CanonicalProjectionDirty{ WorkspaceID: mutation.WorkspaceID, AgentSessionID: mutation.AgentSessionID, RootAgentSessionID: mutation.RootAgentSessionID, RootTurnID: mutation.RootTurnID, MutationID: mutation.MutationID, EntityKind: mutation.EntityKind, EntityID: mutation.EntityID, Operation: mutation.Operation, Version: mutation.Version, }) delta.addView(mutation.WorkspaceID, mutation.AgentSessionID) } return delta } func (delta *CommittedDelta) addView(workspaceID, agentSessionID string) { workspaceID, agentSessionID = strings.TrimSpace(workspaceID), strings.TrimSpace(agentSessionID) if workspaceID == "" || agentSessionID == "" { return } for _, existing := range delta.ViewsInvalidated { if existing.WorkspaceID == workspaceID && existing.AgentSessionID == agentSessionID { return } } delta.ViewsInvalidated = append(delta.ViewsInvalidated, CanonicalViewInvalidated{WorkspaceID: workspaceID, AgentSessionID: agentSessionID}) } func canonicalMessageSessionID(fallback string, messages []storesqlite.Message) string { for _, message := range messages { if value := strings.TrimSpace(message.AgentSessionID); value != "" { return value } } return strings.TrimSpace(fallback) } // NotifyCommitted deliberately swallows observer failures after logging: the // canonical transaction is already committed, and reliable delivery must come // from the durable participant outbox rather than rollback or command failure. func NotifyCommitted(ctx context.Context, observer CommitObserver, delta CommittedDelta) { if observer == nil { return } if err := observer.ObserveCommitted(ctx, delta); err != nil { slog.Warn("agent host commit observer failed", "event", "agent_host.commit_observer.failed", "transaction_id", delta.TransactionID, "error", err, ) } } func (h *Host) notifyCommitted(ctx context.Context, delta CommittedDelta) { if h != nil { h.enrichCommittedDeltaTerminalIdentity(ctx, &delta) ObserveTerminalFailuresFromDelta(ctx, h.terminalFailure, delta) NotifyCommitted(ctx, h.commitObserver, delta) } } type terminalFailureSessionIdentity struct { provider string isChildSession bool } // enrichCommittedDeltaTerminalIdentity resolves session-owned facts from the // canonical store after commit. Runtime and Goal operation rows deliberately // do not duplicate provider identity, while product telemetry still needs the // exact canonical Provider and child-session classification. func (h *Host) enrichCommittedDeltaTerminalIdentity(ctx context.Context, delta *CommittedDelta) { if h == nil || h.store == nil || delta == nil { return } identities := map[string]terminalFailureSessionIdentity{} resolve := func(workspaceID, agentSessionID string) terminalFailureSessionIdentity { workspaceID = strings.TrimSpace(workspaceID) agentSessionID = strings.TrimSpace(agentSessionID) key := workspaceID + "\x00" + agentSessionID if identity, ok := identities[key]; ok { return identity } identity := terminalFailureSessionIdentity{} if workspaceID != "" && agentSessionID != "" { if session, found, err := h.store.GetSession(ctx, workspaceID, agentSessionID); err == nil && found { identity.provider = strings.TrimSpace(session.Provider) identity.isChildSession = canonicalSessionIsChild(session) } } identities[key] = identity return identity } apply := func(provider *string, isChildSession *bool, workspaceID, agentSessionID string) { identity := resolve(workspaceID, agentSessionID) if strings.TrimSpace(*provider) == "" { *provider = identity.provider } *isChildSession = *isChildSession || identity.isChildSession } if committed := delta.RuntimeOperation; committed != nil && committed.Stage == RuntimeOperationFailed { apply(&committed.Provider, &committed.IsChildSession, committed.Operation.WorkspaceID, committed.Operation.AgentSessionID) } if committed := delta.GoalOperation; committed != nil && (committed.Stage == GoalOperationFailed || committed.Stage == GoalOperationTerminal) { workspaceID := firstNonEmptyTrimmed(committed.Operation.WorkspaceID, committed.State.WorkspaceID) agentSessionID := firstNonEmptyTrimmed(committed.Operation.AgentSessionID, committed.State.AgentSessionID) apply(&committed.Provider, &committed.IsChildSession, workspaceID, agentSessionID) } for index := range delta.RootTurnsSettled { settled := &delta.RootTurnsSettled[index] if outcome := strings.TrimSpace(settled.Turn.Outcome); outcome != storesqlite.TurnOutcomeFailed && outcome != storesqlite.TurnOutcomeInterrupted && outcome != storesqlite.TurnOutcomeCanceled { continue } apply(&settled.Provider, &settled.IsChildSession, settled.WorkspaceID, settled.AgentSessionID) } if committed := delta.SessionMessages; committed != nil && len(terminalFailuresFromSessionMessages(*committed, nil)) > 0 { agentSessionID := canonicalMessageSessionID(committed.Input.AgentSessionID, committed.Result.Messages) apply(&committed.Provider, &committed.IsChildSession, committed.Input.WorkspaceID, agentSessionID) } } func canonicalSessionIsChild(session storesqlite.Session) bool { return strings.EqualFold(strings.TrimSpace(session.Kind), storesqlite.SessionKindChild) || strings.TrimSpace(session.ParentToolCallID) != "" }