| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "maps" |
| 9 | "slices" |
| 10 | "strings" |
| 11 | |
| 12 | "reasonix/internal/event" |
| 13 | "reasonix/internal/provider" |
| 14 | ) |
| 15 | |
| 16 | type Projection struct { |
| 17 | Submissions SubmissionIndex |
| 18 | TranscriptInputs []transcriptInput |
| 19 | HiddenTurns map[string]bool |
| 20 | RetractedInputs map[string]string |
| 21 | CommittedSequence uint64 |
| 22 | TurnID string |
| 23 | TurnStatus event.TurnStatus |
| 24 | CurrentTurnStart uint64 |
| 25 | CurrentTurnStartedAt int64 |
| 26 | // CurrentTurnMessageID is the stable identity of the newest assistant |
| 27 | // message committed inside the open turn. It becomes the turn's final reply |
| 28 | // identity when the turn closes. |
| 29 | CurrentTurnMessageID string |
| 30 | CurrentAttempts map[string]bool |
| 31 | CurrentCalls map[string]bool |
| 32 | Turns []TurnBoundary |
| 33 | Messages []provider.Message |
| 34 | // ModelMessages is the exact provider-visible projection. Canonical Messages |
| 35 | // remains the complete UI/history transcript; compaction replaces only this |
| 36 | // view and never deletes the underlying business history. |
| 37 | ModelMessages []provider.Message |
| 38 | Todos []event.Todo |
| 39 | TodoWritten bool |
| 40 | Interactions map[string]string |
| 41 | ActiveTools map[string]string |
| 42 | StartedTools map[string]bool |
| 43 | ActiveSteps map[string]bool |
| 44 | Recovery *event.RecoveryStatus |
| 45 | PlanState json.RawMessage |
| 46 | GoalState json.RawMessage |
| 47 | Title string |
| 48 | // TitleSequence is the sequence of the latest accepted session/title event. |
| 49 | // It is independent from CommittedSequence so ordinary chat appends do not |
| 50 | // conflict with a delayed title mutation. |
| 51 | TitleSequence uint64 |
| 52 | ModelRef string |
| 53 | ModelIdentity string |
| 54 | } |
| 55 | |
| 56 | type TurnBoundary struct { |
| 57 | SamplingCount int `json:"samplingCount,omitempty"` |
| 58 | ToolCount int `json:"toolCount,omitempty"` |
| 59 | DurationMs int64 `json:"durationMs,omitempty"` |
| 60 | TurnID string `json:"turnId"` |
| 61 | StartSequence uint64 `json:"startSequence"` |
| 62 | EndSequence uint64 `json:"endSequence"` |
| 63 | Status event.TurnStatus `json:"status"` |
| 64 | // BoundarySequence is the last sequence of the commit that closed this turn. |
| 65 | // A cut may only land here: a turn end and the state ending with it can share |
| 66 | // one commit, and a cut inside that commit inherits half an operation. |
| 67 | BoundarySequence uint64 `json:"boundarySequence"` |
| 68 | // Availability is fixed from the complete commit that closed the turn. It |
| 69 | // must not be recomputed from the latest projection: a later commit may |
| 70 | // resolve authority that the earlier fork prefix would still inherit. |
| 71 | Availability ForkAvailability `json:"availability"` |
| 72 | // MessageID is the stable transcript identity of the turn's final reply, empty |
| 73 | // when the turn committed none. Surfaces match turns to messages through this |
| 74 | // identity, never through an array position. |
| 75 | MessageID string `json:"messageId,omitempty"` |
| 76 | } |
| 77 | |
| 78 | var ProjectionKinds = map[string]bool{ |
| 79 | "message/complete": true, "message/upsert": true, "message/retract": true, "assistant/attempt": true, |
| 80 | "tool/call": true, "tool/start": true, "tool/result": true, |
| 81 | "turn/start": true, "turn/end": true, "step/start": true, "step/end": true, |
| 82 | "todo/write": true, "interaction/created": true, "interaction/resolved": true, |
| 83 | "plan/state": true, "goal/state": true, "session/title": true, "session/config": true, |
| 84 | "model/context-replace": true, "history/replace": true, |
| 85 | "compaction": true, "runtime/recovery": true, "legacy/import": true, |
| 86 | "diagnostic": true, |
| 87 | } |
| 88 | |
| 89 | var PrototypeProjectionKinds = func() map[string]bool { |
| 90 | kinds := make(map[string]bool, len(ProjectionKinds)+1) |
| 91 | maps.Copy(kinds, ProjectionKinds) |
| 92 | kinds["context/replace"] = true |
| 93 | return kinds |
| 94 | }() |
| 95 | |
| 96 | func Project(commits []Commit) (Projection, error) { |
| 97 | projection := Projection{Todos: []event.Todo{}, Interactions: map[string]string{}, ActiveTools: map[string]string{}} |
| 98 | for _, commit := range commits { |
| 99 | if err := applyProjectionCommit(&projection, commit); err != nil { |
| 100 | return Projection{}, err |
| 101 | } |
| 102 | } |
| 103 | return projection, nil |
| 104 | } |
| 105 | |
| 106 | func applyProjectionCommit(projection *Projection, commit Commit) error { |
| 107 | initializeProjectionMaps(projection) |
| 108 | return applyProjectionEvents(projection, commit) |
| 109 | } |
| 110 | |
| 111 | func initializeProjectionMaps(projection *Projection) { |
| 112 | if projection.Interactions == nil { |
| 113 | projection.Interactions = map[string]string{} |
| 114 | } |
| 115 | if projection.ActiveTools == nil { |
| 116 | projection.ActiveTools = map[string]string{} |
| 117 | } |
| 118 | if projection.StartedTools == nil { |
| 119 | projection.StartedTools = map[string]bool{} |
| 120 | } |
| 121 | if projection.ActiveSteps == nil { |
| 122 | projection.ActiveSteps = map[string]bool{} |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | func applyProjectionEvents(projection *Projection, commit Commit) error { |
| 127 | closedBefore := len(projection.Turns) |
| 128 | for _, ev := range commit.Events { |
| 129 | projection.CommittedSequence = ev.Sequence |
| 130 | var err error |
| 131 | switch ev.Kind { |
| 132 | case "submission/accepted": |
| 133 | err = projectSubmission(projection, commit, ev) |
| 134 | case "legacy/import": |
| 135 | err = projectLegacyImport(projection, commit, ev) |
| 136 | case "message/complete": |
| 137 | err = projectMessageComplete(projection, commit, ev) |
| 138 | case "message/upsert": |
| 139 | err = projectMessageUpsert(projection, commit, ev) |
| 140 | case "message/retract": |
| 141 | err = projectMessageRetract(projection, commit, ev) |
| 142 | case "assistant/attempt": |
| 143 | err = projectAssistantAttempt(projection, commit, ev) |
| 144 | case "history/replace": |
| 145 | err = projectHistoryReplace(projection, commit, ev) |
| 146 | case "model/context-replace": |
| 147 | err = projectModelContextReplace(projection, commit, ev) |
| 148 | case "session/title": |
| 149 | err = projectSessionTitle(projection, commit, ev) |
| 150 | case "session/config": |
| 151 | err = projectSessionConfig(projection, commit, ev) |
| 152 | case "compaction": |
| 153 | err = projectCompaction(projection, commit, ev) |
| 154 | case "turn/start": |
| 155 | err = projectTurnStart(projection, commit, ev) |
| 156 | case "step/start", "step/end": |
| 157 | err = projectStepStart(projection, commit, ev) |
| 158 | case "tool/call": |
| 159 | err = projectToolCall(projection, commit, ev) |
| 160 | case "tool/start": |
| 161 | err = projectToolStart(projection, commit, ev) |
| 162 | case "tool/result": |
| 163 | err = projectToolResult(projection, commit, ev) |
| 164 | case "todo/write": |
| 165 | err = projectTodoWrite(projection, commit, ev) |
| 166 | case "interaction/created": |
| 167 | err = projectInteractionCreated(projection, commit, ev) |
| 168 | case "interaction/resolved": |
| 169 | err = projectInteractionResolved(projection, commit, ev) |
| 170 | case "runtime/recovery": |
| 171 | err = projectRuntimeRecovery(projection, commit, ev) |
| 172 | case "plan/state": |
| 173 | err = projectPlanState(projection, commit, ev) |
| 174 | case "goal/state": |
| 175 | err = projectGoalState(projection, commit, ev) |
| 176 | case "diagnostic": |
| 177 | err = projectDiagnostic(projection, commit, ev) |
| 178 | case "turn/end": |
| 179 | err = projectTurnEnd(projection, commit, ev) |
| 180 | } |
| 181 | if err != nil { |
| 182 | return err |
| 183 | } |
| 184 | applyTranscriptMetadata(projection, commit, ev) |
| 185 | } |
| 186 | // turn/end can be followed by more events in the same atomic commit. Only |
| 187 | // after the whole commit is projected do we know whether its cut leaves a |
| 188 | // turn, interaction, or tool authority open. |
| 189 | for index := closedBefore; index < len(projection.Turns); index++ { |
| 190 | projection.Turns[index].Availability = forkProjectionAvailability(*projection, projection.Turns[index].BoundarySequence) |
| 191 | } |
| 192 | return nil |
| 193 | } |
| 194 | |
| 195 | func projectLegacyImport(projection *Projection, commit Commit, ev Event) error { |
| 196 | var body legacyImportPayload |
| 197 | if err := strictPayload(ev.Payload, &body); err != nil || body.Messages == nil { |
| 198 | return damagedPayload(ev, err) |
| 199 | } |
| 200 | projection.Messages = append([]provider.Message{}, body.Messages...) |
| 201 | projection.ModelMessages = append([]provider.Message{}, provider.ModelMessages(body.Messages)...) |
| 202 | projection.GoalState = cloneRaw(body.Goal) |
| 203 | projection.ModelRef = strings.TrimSpace(body.ModelRef) |
| 204 | projection.ModelIdentity = strings.TrimSpace(body.ModelIdentity) |
| 205 | return nil |
| 206 | } |
| 207 | |
| 208 | func projectMessageComplete(projection *Projection, commit Commit, ev Event) error { |
| 209 | var body struct { |
| 210 | Message *provider.Message `json:"message"` |
| 211 | } |
| 212 | if err := strictPayload(ev.Payload, &body); err != nil || body.Message == nil || body.Message.ID == "" { |
| 213 | return damagedPayload(ev, err) |
| 214 | } |
| 215 | if projectionMessageIndex(projection.Messages, body.Message.ID) >= 0 { |
| 216 | return damagedPayload(ev, fmt.Errorf("duplicate stable message id %q", body.Message.ID)) |
| 217 | } |
| 218 | projection.Messages = append(projection.Messages, *body.Message) |
| 219 | projection.ModelMessages = append(projection.ModelMessages, provider.ModelMessages([]provider.Message{*body.Message})...) |
| 220 | projection.recordTurnReply(*body.Message) |
| 221 | return nil |
| 222 | } |
| 223 | |
| 224 | // recordTurnReply keeps the open turn's final answer identity. A fork entry |
| 225 | // belongs on the turn's answer, so a trailing tool call, a retried attempt, or a |
| 226 | // host-generated protocol message must not take the anchor away from the text a |
| 227 | // user actually reads. |
| 228 | func (projection *Projection) recordTurnReply(message provider.Message) { |
| 229 | if projection.TurnID == "" || message.Role != provider.RoleAssistant || message.LocalOnly { |
| 230 | return |
| 231 | } |
| 232 | if strings.TrimSpace(message.RawContent) == "" && strings.TrimSpace(message.Content) == "" { |
| 233 | return |
| 234 | } |
| 235 | projection.CurrentTurnMessageID = message.ID |
| 236 | } |
| 237 | |
| 238 | func projectMessageUpsert(projection *Projection, commit Commit, ev Event) error { |
| 239 | var body struct { |
| 240 | Message *provider.Message `json:"message"` |
| 241 | } |
| 242 | if err := strictPayload(ev.Payload, &body); err != nil || body.Message == nil || body.Message.ID == "" { |
| 243 | return damagedPayload(ev, err) |
| 244 | } |
| 245 | if !replaceProjectionMessage(projection.Messages, *body.Message) { |
| 246 | projection.Messages = append(projection.Messages, *body.Message) |
| 247 | } |
| 248 | visible := provider.ModelMessages([]provider.Message{*body.Message}) |
| 249 | modelIndex := projectionMessageIndex(projection.ModelMessages, body.Message.ID) |
| 250 | if modelIndex >= 0 { |
| 251 | if len(visible) == 0 { |
| 252 | projection.ModelMessages = append(projection.ModelMessages[:modelIndex], projection.ModelMessages[modelIndex+1:]...) |
| 253 | } else { |
| 254 | projection.ModelMessages[modelIndex] = visible[0] |
| 255 | } |
| 256 | } else if len(visible) > 0 { |
| 257 | // Upserts are normally metadata changes to an existing message. The |
| 258 | // append case is retained for explicitly-created records. |
| 259 | projection.ModelMessages = append(projection.ModelMessages, visible[0]) |
| 260 | } |
| 261 | if projection.CurrentTurnMessageID == body.Message.ID && (body.Message.LocalOnly || body.Message.Role != provider.RoleAssistant) { |
| 262 | projection.CurrentTurnMessageID = "" |
| 263 | } |
| 264 | projection.recordTurnReply(*body.Message) |
| 265 | return nil |
| 266 | } |
| 267 | |
| 268 | func projectMessageRetract(projection *Projection, commit Commit, ev Event) error { |
| 269 | ids, err := retractedMessageIDs(ev, ev.Payload) |
| 270 | if err != nil { |
| 271 | return err |
| 272 | } |
| 273 | removed := make(map[string]bool, len(ids)) |
| 274 | for _, id := range ids { |
| 275 | removed[id] = true |
| 276 | } |
| 277 | filter := func(messages []provider.Message) []provider.Message { |
| 278 | return slices.DeleteFunc(messages, func(message provider.Message) bool { return removed[message.ID] }) |
| 279 | } |
| 280 | projection.Messages = filter(projection.Messages) |
| 281 | projection.ModelMessages = filter(projection.ModelMessages) |
| 282 | if removed[projection.CurrentTurnMessageID] { |
| 283 | projection.CurrentTurnMessageID = "" |
| 284 | } |
| 285 | return nil |
| 286 | } |
| 287 | |
| 288 | func projectAssistantAttempt(projection *Projection, commit Commit, ev Event) error { |
| 289 | var body struct { |
| 290 | ID string `json:"id"` |
| 291 | MessageID string `json:"messageId,omitempty"` |
| 292 | Action string `json:"action"` |
| 293 | Attempt int `json:"attempt,omitempty"` |
| 294 | Max int `json:"max,omitempty"` |
| 295 | Reason string `json:"reason,omitempty"` |
| 296 | } |
| 297 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" || (body.Action != "begin" && body.Action != "discard" && body.Action != "commit") { |
| 298 | return damagedPayload(ev, err) |
| 299 | } |
| 300 | if projection.TurnID != "" && body.Action == "begin" { |
| 301 | if projection.CurrentAttempts == nil { |
| 302 | projection.CurrentAttempts = map[string]bool{} |
| 303 | } |
| 304 | projection.CurrentAttempts[body.ID] = true |
| 305 | } |
| 306 | return nil |
| 307 | } |
| 308 | |
| 309 | func projectHistoryReplace(projection *Projection, commit Commit, ev Event) error { |
| 310 | var body historyReplacePayload |
| 311 | if err := strictPayload(ev.Payload, &body); err != nil || body.Messages == nil { |
| 312 | return damagedPayload(ev, err) |
| 313 | } |
| 314 | projection.Messages = append([]provider.Message(nil), body.Messages...) |
| 315 | projection.ModelMessages = append([]provider.Message(nil), provider.ModelMessages(body.Messages)...) |
| 316 | return nil |
| 317 | } |
| 318 | |
| 319 | func projectModelContextReplace(projection *Projection, commit Commit, ev Event) error { |
| 320 | var body historyReplacePayload |
| 321 | if err := strictPayload(ev.Payload, &body); err != nil || body.Messages == nil { |
| 322 | return damagedPayload(ev, err) |
| 323 | } |
| 324 | projection.ModelMessages = append([]provider.Message(nil), body.Messages...) |
| 325 | return nil |
| 326 | } |
| 327 | |
| 328 | func projectSessionTitle(projection *Projection, commit Commit, ev Event) error { |
| 329 | var body struct { |
| 330 | Title string `json:"title"` |
| 331 | } |
| 332 | if err := strictPayload(ev.Payload, &body); err != nil { |
| 333 | return damagedPayload(ev, err) |
| 334 | } |
| 335 | projection.Title = body.Title |
| 336 | // Sequence zero is a valid first event, so store the one-based identity; |
| 337 | // zero remains the durable "no title event yet" revision. |
| 338 | projection.TitleSequence = ev.Sequence + 1 |
| 339 | return nil |
| 340 | } |
| 341 | |
| 342 | func projectSessionConfig(projection *Projection, commit Commit, ev Event) error { |
| 343 | var body struct { |
| 344 | ModelRef string `json:"modelRef"` |
| 345 | ModelIdentity string `json:"modelIdentity,omitempty"` |
| 346 | } |
| 347 | if err := strictPayload(ev.Payload, &body); err != nil || strings.TrimSpace(body.ModelRef) == "" { |
| 348 | return damagedPayload(ev, err) |
| 349 | } |
| 350 | projection.ModelRef = strings.TrimSpace(body.ModelRef) |
| 351 | projection.ModelIdentity = strings.TrimSpace(body.ModelIdentity) |
| 352 | return nil |
| 353 | } |
| 354 | |
| 355 | func projectCompaction(projection *Projection, commit Commit, ev Event) error { |
| 356 | var body struct { |
| 357 | Messages []provider.Message `json:"messages"` |
| 358 | Trigger string `json:"trigger,omitempty"` |
| 359 | Sources []uint64 `json:"sourceSequences,omitempty"` |
| 360 | } |
| 361 | if err := strictPayload(ev.Payload, &body); err != nil || body.Messages == nil { |
| 362 | return damagedPayload(ev, err) |
| 363 | } |
| 364 | projection.ModelMessages = append([]provider.Message(nil), body.Messages...) |
| 365 | return nil |
| 366 | } |
| 367 | |
| 368 | func projectTurnStart(projection *Projection, commit Commit, ev Event) error { |
| 369 | projection.TurnID = commit.TurnID |
| 370 | projection.TurnStatus = event.TurnInProgress |
| 371 | projection.CurrentTurnStart = ev.Sequence |
| 372 | projection.CurrentTurnStartedAt = commit.CreatedAt.UnixMilli() |
| 373 | projection.CurrentTurnMessageID = "" |
| 374 | projection.CurrentAttempts, projection.CurrentCalls = map[string]bool{}, map[string]bool{} |
| 375 | projection.Todos, projection.TodoWritten = []event.Todo{}, false |
| 376 | projection.Recovery = nil |
| 377 | return nil |
| 378 | } |
| 379 | |
| 380 | func projectStepStart(projection *Projection, commit Commit, ev Event) error { |
| 381 | var body struct { |
| 382 | ID string `json:"id"` |
| 383 | Status string `json:"status,omitempty"` |
| 384 | } |
| 385 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" { |
| 386 | return damagedPayload(ev, err) |
| 387 | } |
| 388 | if ev.Kind == "step/start" { |
| 389 | projection.ActiveSteps[body.ID] = true |
| 390 | } else { |
| 391 | delete(projection.ActiveSteps, body.ID) |
| 392 | } |
| 393 | return nil |
| 394 | } |
| 395 | |
| 396 | func projectToolCall(projection *Projection, commit Commit, ev Event) error { |
| 397 | var body struct { |
| 398 | ID string `json:"id"` |
| 399 | Name string `json:"name"` |
| 400 | Args string `json:"args,omitempty"` |
| 401 | RunState provider.ToolRunState `json:"runState,omitempty"` |
| 402 | Diagnostic json.RawMessage `json:"diagnostic,omitempty"` |
| 403 | ResolvedName string `json:"resolvedName,omitempty"` |
| 404 | CapabilityID string `json:"capabilityId,omitempty"` |
| 405 | ReadOnly bool `json:"readOnly,omitempty"` |
| 406 | Truncated bool `json:"truncated,omitempty"` |
| 407 | DurationMs int64 `json:"durationMs,omitempty"` |
| 408 | StartedAt int64 `json:"startedAt,omitempty"` |
| 409 | EndedAt int64 `json:"endedAt,omitempty"` |
| 410 | Partial bool `json:"partial,omitempty"` |
| 411 | ArgChars int `json:"argChars,omitempty"` |
| 412 | Refreshed bool `json:"refreshed,omitempty"` |
| 413 | ParentID string `json:"parentId,omitempty"` |
| 414 | AttemptID string `json:"attemptId,omitempty"` |
| 415 | SubagentRef string `json:"subagentRef,omitempty"` |
| 416 | SubagentStatus string `json:"subagentStatus,omitempty"` |
| 417 | SubagentErrorCode string `json:"subagentErrorCode,omitempty"` |
| 418 | SubagentRetryable bool `json:"subagentRetryable,omitempty"` |
| 419 | Diff string `json:"diff,omitempty"` |
| 420 | Added int `json:"added,omitempty"` |
| 421 | Removed int `json:"removed,omitempty"` |
| 422 | Profile json.RawMessage `json:"profile,omitempty"` |
| 423 | Execution json.RawMessage `json:"execution,omitempty"` |
| 424 | PresentedFiles []provider.PresentedFile `json:"presentedFiles,omitempty"` |
| 425 | WorkspaceMutation bool `json:"workspaceMutation,omitempty"` |
| 426 | WorkspacePaths []string `json:"workspacePaths,omitempty"` |
| 427 | WorkspaceAllPaths bool `json:"workspaceAllPaths,omitempty"` |
| 428 | } |
| 429 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" || body.Name == "" { |
| 430 | return damagedPayload(ev, err) |
| 431 | } |
| 432 | projection.ActiveTools[body.ID] = body.Name |
| 433 | if projection.TurnID != "" { |
| 434 | if projection.CurrentCalls == nil { |
| 435 | projection.CurrentCalls = map[string]bool{} |
| 436 | } |
| 437 | projection.CurrentCalls[body.ID] = true |
| 438 | } |
| 439 | return nil |
| 440 | } |
| 441 | |
| 442 | func projectToolStart(projection *Projection, commit Commit, ev Event) error { |
| 443 | var body struct { |
| 444 | ID string `json:"id"` |
| 445 | Name string `json:"name"` |
| 446 | } |
| 447 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" || body.Name == "" { |
| 448 | return damagedPayload(ev, err) |
| 449 | } |
| 450 | projection.ActiveTools[body.ID] = body.Name |
| 451 | projection.StartedTools[body.ID] = true |
| 452 | return nil |
| 453 | } |
| 454 | |
| 455 | func projectToolResult(projection *Projection, commit Commit, ev Event) error { |
| 456 | var body struct { |
| 457 | ID string `json:"id"` |
| 458 | Name string `json:"name"` |
| 459 | Args string `json:"args,omitempty"` |
| 460 | Error string `json:"error,omitempty"` |
| 461 | Output string `json:"output,omitempty"` |
| 462 | State string `json:"state,omitempty"` |
| 463 | RunState provider.ToolRunState `json:"runState,omitempty"` |
| 464 | Diagnostic json.RawMessage `json:"diagnostic,omitempty"` |
| 465 | ResolvedName string `json:"resolvedName,omitempty"` |
| 466 | CapabilityID string `json:"capabilityId,omitempty"` |
| 467 | ReadOnly bool `json:"readOnly,omitempty"` |
| 468 | Truncated bool `json:"truncated,omitempty"` |
| 469 | DurationMs int64 `json:"durationMs,omitempty"` |
| 470 | StartedAt int64 `json:"startedAt,omitempty"` |
| 471 | EndedAt int64 `json:"endedAt,omitempty"` |
| 472 | Partial bool `json:"partial,omitempty"` |
| 473 | ArgChars int `json:"argChars,omitempty"` |
| 474 | Refreshed bool `json:"refreshed,omitempty"` |
| 475 | ParentID string `json:"parentId,omitempty"` |
| 476 | AttemptID string `json:"attemptId,omitempty"` |
| 477 | SubagentRef string `json:"subagentRef,omitempty"` |
| 478 | SubagentStatus string `json:"subagentStatus,omitempty"` |
| 479 | SubagentErrorCode string `json:"subagentErrorCode,omitempty"` |
| 480 | SubagentRetryable bool `json:"subagentRetryable,omitempty"` |
| 481 | Diff string `json:"diff,omitempty"` |
| 482 | Added int `json:"added,omitempty"` |
| 483 | Removed int `json:"removed,omitempty"` |
| 484 | Profile json.RawMessage `json:"profile,omitempty"` |
| 485 | Execution json.RawMessage `json:"execution,omitempty"` |
| 486 | PresentedFiles []provider.PresentedFile `json:"presentedFiles,omitempty"` |
| 487 | Todos []event.Todo `json:"todos,omitempty"` |
| 488 | TodoWritten bool `json:"todoWritten,omitempty"` |
| 489 | WorkspaceMutation bool `json:"workspaceMutation,omitempty"` |
| 490 | WorkspacePaths []string `json:"workspacePaths,omitempty"` |
| 491 | WorkspaceAllPaths bool `json:"workspaceAllPaths,omitempty"` |
| 492 | } |
| 493 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" || body.Name == "" { |
| 494 | return damagedPayload(ev, err) |
| 495 | } |
| 496 | delete(projection.ActiveTools, body.ID) |
| 497 | delete(projection.StartedTools, body.ID) |
| 498 | return nil |
| 499 | } |
| 500 | |
| 501 | func projectTodoWrite(projection *Projection, commit Commit, ev Event) error { |
| 502 | var body struct { |
| 503 | Todos []event.Todo `json:"todos"` |
| 504 | } |
| 505 | if err := strictPayload(ev.Payload, &body); err != nil || validateTodos(body.Todos) != nil { |
| 506 | return damagedPayload(ev, err) |
| 507 | } |
| 508 | projection.Todos, projection.TodoWritten = append([]event.Todo(nil), body.Todos...), true |
| 509 | return nil |
| 510 | } |
| 511 | |
| 512 | func projectInteractionCreated(projection *Projection, commit Commit, ev Event) error { |
| 513 | var body struct { |
| 514 | ID string `json:"id"` |
| 515 | ToolCallID string `json:"toolCallId,omitempty"` |
| 516 | Kind string `json:"kind,omitempty"` |
| 517 | State string `json:"state,omitempty"` |
| 518 | SessionID string `json:"sessionId,omitempty"` |
| 519 | HeadID string `json:"headId,omitempty"` |
| 520 | TurnID string `json:"turnId,omitempty"` |
| 521 | RuntimeEpoch string `json:"runtimeEpoch,omitempty"` |
| 522 | } |
| 523 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" || (body.State != "" && body.State != "pending") { |
| 524 | return damagedPayload(ev, err) |
| 525 | } |
| 526 | projection.Interactions[body.ID] = "pending" |
| 527 | return nil |
| 528 | } |
| 529 | |
| 530 | func projectInteractionResolved(projection *Projection, commit Commit, ev Event) error { |
| 531 | var body struct { |
| 532 | ID string `json:"id"` |
| 533 | State string `json:"state"` |
| 534 | } |
| 535 | if err := strictPayload(ev.Payload, &body); err != nil || body.ID == "" || !terminalInteractionState(body.State) { |
| 536 | return damagedPayload(ev, err) |
| 537 | } |
| 538 | delete(projection.Interactions, body.ID) |
| 539 | return nil |
| 540 | } |
| 541 | |
| 542 | func projectRuntimeRecovery(projection *Projection, commit Commit, ev Event) error { |
| 543 | var body event.RecoveryStatus |
| 544 | if err := strictPayload(ev.Payload, &body); err != nil { |
| 545 | return damagedPayload(ev, err) |
| 546 | } |
| 547 | projection.Recovery = &body |
| 548 | return nil |
| 549 | } |
| 550 | |
| 551 | func projectPlanState(projection *Projection, commit Commit, ev Event) error { |
| 552 | if !validJSONObject(ev.Payload) { |
| 553 | return damagedPayload(ev, nil) |
| 554 | } |
| 555 | projection.PlanState = cloneRaw(ev.Payload) |
| 556 | return nil |
| 557 | } |
| 558 | |
| 559 | func projectGoalState(projection *Projection, commit Commit, ev Event) error { |
| 560 | if !validJSONObject(ev.Payload) { |
| 561 | return damagedPayload(ev, nil) |
| 562 | } |
| 563 | projection.GoalState = cloneRaw(ev.Payload) |
| 564 | return nil |
| 565 | } |
| 566 | |
| 567 | func projectDiagnostic(projection *Projection, commit Commit, ev Event) error { |
| 568 | if len(ev.Payload) > 0 && !json.Valid(ev.Payload) { |
| 569 | return damagedPayload(ev, nil) |
| 570 | } |
| 571 | return nil |
| 572 | } |
| 573 | |
| 574 | func projectTurnEnd(projection *Projection, commit Commit, ev Event) error { |
| 575 | var body struct { |
| 576 | Status event.TurnStatus `json:"status"` |
| 577 | } |
| 578 | if err := strictPayload(ev.Payload, &body); err != nil || !body.Status.Terminal() { |
| 579 | return damagedPayload(ev, err) |
| 580 | } |
| 581 | if projection.TurnID != "" && projection.CurrentTurnStart != 0 { |
| 582 | projection.Turns = append(projection.Turns, TurnBoundary{ |
| 583 | SamplingCount: len(projection.CurrentAttempts), ToolCount: len(projection.CurrentCalls), |
| 584 | DurationMs: max(0, commit.CreatedAt.UnixMilli()-projection.CurrentTurnStartedAt), |
| 585 | TurnID: projection.TurnID, StartSequence: projection.CurrentTurnStart, |
| 586 | EndSequence: ev.Sequence, Status: body.Status, |
| 587 | BoundarySequence: commit.LastSequence(), |
| 588 | MessageID: projection.CurrentTurnMessageID, |
| 589 | }) |
| 590 | } |
| 591 | projection.TurnID = "" |
| 592 | projection.CurrentTurnStart = 0 |
| 593 | projection.CurrentTurnStartedAt = 0 |
| 594 | projection.CurrentTurnMessageID = "" |
| 595 | projection.TurnStatus = body.Status |
| 596 | return nil |
| 597 | } |
| 598 | |
| 599 | func replaceProjectionMessage(messages []provider.Message, replacement provider.Message) bool { |
| 600 | if index := projectionMessageIndex(messages, replacement.ID); index >= 0 { |
| 601 | messages[index] = replacement |
| 602 | return true |
| 603 | } |
| 604 | return false |
| 605 | } |
| 606 | |
| 607 | func projectionMessageIndex(messages []provider.Message, id string) int { |
| 608 | for i := range messages { |
| 609 | if messages[i].ID == id { |
| 610 | return i |
| 611 | } |
| 612 | } |
| 613 | return -1 |
| 614 | } |
| 615 | |
| 616 | func cloneProjection(projection Projection) Projection { |
| 617 | projection.TranscriptInputs = append([]transcriptInput(nil), projection.TranscriptInputs...) |
| 618 | projection.HiddenTurns = maps.Clone(projection.HiddenTurns) |
| 619 | projection.RetractedInputs = maps.Clone(projection.RetractedInputs) |
| 620 | projection.CurrentAttempts = maps.Clone(projection.CurrentAttempts) |
| 621 | projection.CurrentCalls = maps.Clone(projection.CurrentCalls) |
| 622 | projection.Messages = append([]provider.Message(nil), projection.Messages...) |
| 623 | projection.ModelMessages = append([]provider.Message(nil), projection.ModelMessages...) |
| 624 | projection.Turns = append([]TurnBoundary(nil), projection.Turns...) |
| 625 | projection.Todos = append([]event.Todo(nil), projection.Todos...) |
| 626 | interactions := make(map[string]string, len(projection.Interactions)) |
| 627 | maps.Copy(interactions, projection.Interactions) |
| 628 | projection.Interactions = interactions |
| 629 | tools := make(map[string]string, len(projection.ActiveTools)) |
| 630 | maps.Copy(tools, projection.ActiveTools) |
| 631 | projection.ActiveTools = tools |
| 632 | projection.StartedTools = maps.Clone(projection.StartedTools) |
| 633 | projection.ActiveSteps = maps.Clone(projection.ActiveSteps) |
| 634 | if projection.Recovery != nil { |
| 635 | recovery := *projection.Recovery |
| 636 | projection.Recovery = &recovery |
| 637 | } |
| 638 | projection.PlanState = cloneRaw(projection.PlanState) |
| 639 | projection.GoalState = cloneRaw(projection.GoalState) |
| 640 | return projection |
| 641 | } |
| 642 | |
| 643 | func strictPayload(payload json.RawMessage, target any) error { |
| 644 | if len(payload) == 0 { |
| 645 | return io.ErrUnexpectedEOF |
| 646 | } |
| 647 | decoder := json.NewDecoder(bytes.NewReader(payload)) |
| 648 | decoder.DisallowUnknownFields() |
| 649 | if err := decoder.Decode(target); err != nil { |
| 650 | return err |
| 651 | } |
| 652 | var extra any |
| 653 | if err := decoder.Decode(&extra); err != io.EOF { |
| 654 | if err == nil { |
| 655 | return fmt.Errorf("multiple JSON values") |
| 656 | } |
| 657 | return err |
| 658 | } |
| 659 | return nil |
| 660 | } |
| 661 | |
| 662 | func validateTodos(todos []event.Todo) error { |
| 663 | if todos == nil { |
| 664 | return fmt.Errorf("todos must be an array") |
| 665 | } |
| 666 | seen := make(map[string]bool, len(todos)) |
| 667 | for i, todo := range todos { |
| 668 | content := strings.TrimSpace(todo.Content) |
| 669 | if content == "" || content != todo.Content || seen[content] { |
| 670 | return fmt.Errorf("todos[%d].content is invalid", i) |
| 671 | } |
| 672 | seen[content] = true |
| 673 | switch todo.Status { |
| 674 | case "pending", "in_progress", "completed": |
| 675 | default: |
| 676 | return fmt.Errorf("todos[%d].status is invalid", i) |
| 677 | } |
| 678 | } |
| 679 | return nil |
| 680 | } |
| 681 | |
| 682 | func terminalInteractionState(state string) bool { |
| 683 | switch state { |
| 684 | case "answered", "rejected", "cancelled", "unavailable": |
| 685 | return true |
| 686 | default: |
| 687 | return false |
| 688 | } |
| 689 | } |
| 690 | |
| 691 | func validJSONObject(raw json.RawMessage) bool { |
| 692 | var object map[string]json.RawMessage |
| 693 | return len(raw) > 0 && json.Unmarshal(raw, &object) == nil && object != nil |
| 694 | } |
| 695 | |
| 696 | func damagedPayload(event Event, cause error) error { |
| 697 | if cause != nil { |
| 698 | return fmt.Errorf("%w: invalid %s payload at %d: %w", ErrDamagedStore, event.Kind, event.Sequence, cause) |
| 699 | } |
| 700 | return fmt.Errorf("%w: invalid %s payload at %d", ErrDamagedStore, event.Kind, event.Sequence) |
| 701 | } |
| 702 | |
| 703 | func cloneRaw(raw json.RawMessage) json.RawMessage { return append(json.RawMessage(nil), raw...) } |
| 704 |