package api import ( "encoding/json" "fmt" "log" "net/http" "strings" "time" "github.com/google/uuid" "crypto-miner-server/internal/ai" dbpkg "crypto-miner-server/internal/db" "crypto-miner-server/internal/strategy" ) // 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 } hospice := map[string]bool{} if h.hub != nil { hospice = h.hub.HospiceStrainSet() } var spawned []*TimelineBranch for _, persona := range normalized { lanes := ai.PersonaSpreadTierOrder(persona) if len(lanes) > 0 && strategy.LaneInHospice(lanes[0], hospice) { continue } 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) } if len(spawned) == 0 { h.mu.Unlock() http.Error(w, "all fork personas map to hospice strains — parent selection blocked", http.StatusConflict) return } h.mu.Unlock() h.persistSession(sess) for _, branch := range spawned { b := branch go h.runGhostBranch(sess.ID, b) } h.broadcastTimelineEvent(sess, "fork", spawned[0]) if h.hub != nil && h.hub.db != nil { why := map[string]interface{}{ "session_id": sess.ID, "fork_hop_index": req.ForkHopIndex, "target_agent_id": target.AgentID, "personas": normalized, "branches_spawned": len(spawned), } _ = (&OathLedgerBridge{DB: h.hub.db, Hub: h.hub}).Record( AuthUsername(r), dbpkg.OathForkMerge, target.AgentID, agentStrainFromDB(h.hub.db, target.AgentID), dbpkg.OathOutcomePending, why, map[string]interface{}{ "event": "fork", "session_id": sess.ID, "fork_hop_index": req.ForkHopIndex, "branches_spawned": len(spawned), }, ) } 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 h.hub != nil && len(winner.SpreadLanes) > 0 { hospice := h.hub.HospiceStrainSet() if strategy.LaneInHospice(winner.SpreadLanes[0], hospice) { h.mu.Unlock() http.Error(w, "branch spread strain is in hospice — fork-merge parent selection blocked", http.StatusConflict) 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) if h.hub != nil && h.hub.db != nil { why := map[string]interface{}{ "session_id": sess.ID, "merged_branch_id": winner.ID, "merged_persona": winner.Persona, "merged_spread_lane": sess.MergedSpreadLane, "merged_hashrate": sess.MergedHashrate, } _ = (&OathLedgerBridge{DB: h.hub.db, Hub: h.hub}).Record( AuthUsername(r), dbpkg.OathForkMerge, winner.TargetAgentID, agentStrainFromDB(h.hub.db, winner.TargetAgentID), dbpkg.OathOutcomeSuccess, why, map[string]interface{}{ "event": "merge", "session_id": sess.ID, "merged_branch_id": winner.ID, "merged_persona": winner.Persona, "merged_spread_lane": sess.MergedSpreadLane, }, ) } 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 }