| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "reasonix/internal/sessioncontent" |
| 7 | ) |
| 8 | |
| 9 | // ToolObservation separates execution evidence from resident result bodies. |
| 10 | type ToolObservation struct { |
| 11 | State string `json:"state"` |
| 12 | MessageID string `json:"messageId,omitempty"` |
| 13 | Version int `json:"version,omitempty"` |
| 14 | ContentRef *sessioncontent.Ref `json:"contentRef,omitempty"` |
| 15 | } |
| 16 | |
| 17 | func (q *Query) attachToolObservations(ctx context.Context, db *sql.DB, ref SessionRef, page *HistoryWindowPage) error { |
| 18 | for i := range page.Messages { |
| 19 | message := &page.Messages[i] |
| 20 | if message.Role != "assistant" { |
| 21 | continue |
| 22 | } |
| 23 | rows, err := db.QueryContext(ctx, `SELECT calls.call_id,COALESCE(CASE WHEN result.message_id IS NOT NULL THEN results.state END,(SELECT state FROM tool_states WHERE call_id=calls.call_id AND sequence<=? ORDER BY sequence DESC LIMIT 1),'unknown'),COALESCE(result.message_id,''),COALESCE(result.version,0),COALESCE(result.content_digest,''),COALESCE(result.content_bytes,0),COALESCE(result.content_index_digest,'') |
| 24 | FROM messages owner JOIN tool_links calls ON calls.message_id=owner.message_id AND calls.digest=owner.content_digest AND calls.is_result=0 |
| 25 | LEFT JOIN tool_links results ON results.call_id=calls.call_id AND results.is_result=1 |
| 26 | LEFT JOIN messages result ON result.message_id=results.message_id AND result.content_digest=results.digest AND result.event_sequence<=? AND (result.valid_to=0 OR result.valid_to>?) |
| 27 | WHERE owner.message_id=? AND owner.version=? ORDER BY result.event_sequence DESC`, page.SnapshotSequence, page.SnapshotSequence, page.SnapshotSequence, message.MessageID, message.Version) |
| 28 | if err != nil { |
| 29 | return err |
| 30 | } |
| 31 | observations := map[string]ToolObservation{} |
| 32 | for rows.Next() { |
| 33 | var id, digest, index string |
| 34 | var size int64 |
| 35 | var observation ToolObservation |
| 36 | if err = rows.Scan(&id, &observation.State, &observation.MessageID, &observation.Version, &digest, &size, &index); err != nil { |
| 37 | rows.Close() |
| 38 | return err |
| 39 | } |
| 40 | if prior, ok := observations[id]; ok && prior.MessageID != "" { |
| 41 | continue |
| 42 | } |
| 43 | if observation.MessageID != "" && digest != "" { |
| 44 | observation.ContentRef = &sessioncontent.Ref{Digest: digest, Bytes: size, IndexDigest: index, IntegrityBlock: sessioncontent.IntegrityBlockBytes, MediaType: "application/json"} |
| 45 | q.authorizeContentForGeneration(ref.SessionID, q.storageGeneration(ref.SessionID), digest, size, index) |
| 46 | } |
| 47 | observations[id] = observation |
| 48 | } |
| 49 | err = rows.Err() |
| 50 | rows.Close() |
| 51 | if err != nil { |
| 52 | return err |
| 53 | } |
| 54 | if len(observations) > 0 { |
| 55 | message.ToolObservations = observations |
| 56 | } |
| 57 | } |
| 58 | return nil |
| 59 | } |
| 60 |