#11 test refactors and minor changes in both golang and python sdk wi…

README.md

@@ -292,7 +292,7 @@ For non-Python clients (Next.js, n8n, curl, Slack bots).
 |---|---|
 | `GET /api/workers` | Live list of every worker on the mesh with telemetry. |
 | `POST /api/execute` | Sync streaming task dispatch. Body: `{"worker_id":..., "prompt":..., "task_id":...}`. Streams worker stdout back chunked. |
-| `POST /api/execute/async` | Fire-and-forget. Returns `202 {"task_id":...}` immediately. POSTs to `webhook_url` on completion (signed if `AGENTFM_WEBHOOK_SECRET` is set). |
+| `POST /api/execute/async` | Fire-and-forget. Returns `202 {"task_id":...}` immediately. POSTs to `webhook_url` on completion (signed if `AGENTFM_WEBHOOK_SECRET` is set). The background task spawns *before* the 202 ack is written: a 202 means the task is being executed, even if the client hangs up before reading the body. |
 | `GET /metrics` | Prometheus scrape endpoint (see Observability). |
 
 Webhook POSTs are bounded by a 30 s timeout, do not follow redirects, and the response body is capped at 64 KiB.

agentfm-go/cmd/agentfm/main.go

@@ -76,13 +76,20 @@ func main() {
 		return
 	}
 
-	validateOperatorConfig(cfg)
-
 	if *mode == "" {
 		pterm.Error.Println("Please specify a mode: -mode boss, worker, relay, api, test, or genkey")
 		os.Exit(1)
 	}
 
+	// Worker-config bounds (maxtasks, maxcpu, maxgpu, agent/desc/model
+	// lengths) only apply to roles that actually consume worker.Config.
+	// Boss/relay/api take defaults, so running the validator there would
+	// just be noise — and would surface confusing limits in --help that
+	// don't apply to the chosen mode.
+	if *mode == "worker" || *mode == "test" {
+		validateOperatorConfig(cfg)
+	}
+
 	ctx := context.Background()
 
 	netCfg := network.Config{

agentfm-go/cmd/relay/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"log/slog"
 	"os"
 	"os/signal"
 	"syscall"
@@ -27,25 +28,12 @@ func fatalf(format string, args ...interface{}) {
 	pterm.Fatal.Printfln(format, args...)
 }
 
-// This function loads a saved identity or generates a new one
+// getStaticIdentity is a thin wrapper around network.LoadOrGenerateIdentity
+// kept for the cmd/relay binary's existing logging idiom. The shared helper
+// takes care of the corrupt-file warning, the 0600 perm, and the
+// regenerate-on-missing semantics so both binaries stay in lockstep.
 func getStaticIdentity(keyFile string) (crypto.PrivKey, error) {
-	if keyBytes, err := os.ReadFile(keyFile); err == nil {
-		fmt.Printf("🔑 Loaded existing permanent identity from %s!\n", keyFile)
-		return crypto.UnmarshalPrivateKey(keyBytes)
-	}
-
-	fmt.Printf("✨ Generating new permanent identity at %s...\n", keyFile)
-	priv, _, err := crypto.GenerateKeyPair(crypto.Ed25519, -1)
-	if err != nil {
-		return nil, err
-	}
-
-	keyBytes, err := crypto.MarshalPrivateKey(priv)
-	if err != nil {
-		return nil, err
-	}
-	err = os.WriteFile(keyFile, keyBytes, 0600)
-	return priv, err
+	return network.LoadOrGenerateIdentity(keyFile)
 }
 
 func main() {
@@ -166,6 +154,10 @@ func main() {
 
 	fmt.Println("\nShutting down relay node...")
 	if err := host.Close(); err != nil {
-		pterm.Error.Printfln("Host close error: %v", err)
+		// slog (not pterm) so log shippers see structured failure events
+		// for relays running unattended under systemd / docker / k8s.
+		slog.Error("relay host close",
+			slog.Any(obs.FieldErr, err),
+		)
 	}
 }

agentfm-go/internal/boss/api_async.go

@@ -38,6 +38,17 @@ const asyncArtifactWait = 10 * time.Second
 // Returns whether the file was observed (currently informational only;
 // callers fire the webhook either way).
 func waitForArtifact(ctx context.Context, taskID string, maxWait time.Duration) bool {
+	// Defense-in-depth: today the only caller passes a newCompletionID()
+	// (crypto/rand hex), so a path-traversal payload is impossible. But
+	// the helper is generic; refuse anything that can't safely be joined
+	// into a filesystem path so a future caller can't accidentally
+	// inherit a traversal bug.
+	if !network.SafeTaskIDPattern.MatchString(taskID) {
+		slog.Warn("waitForArtifact rejected unsafe taskID",
+			slog.String(obs.FieldTaskID, taskID),
+		)
+		return false
+	}
 	deadline := time.Now().Add(maxWait)
 	zipPath := filepath.Join("agentfm_artifacts", taskID+".zip")
 	ticker := time.NewTicker(100 * time.Millisecond)
@@ -107,23 +118,13 @@ func (b *Boss) asyncExecuteHandler(rootCtx context.Context, wg *sync.WaitGroup)
 			return
 		}
 
-		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(http.StatusAccepted)
-		if err := json.NewEncoder(w).Encode(map[string]string{
-			"task_id": taskID,
-			"status":  "queued",
-			"message": "Task dispatched to P2P mesh.",
-		}); err != nil {
-			slog.Error("async task ack write", slog.Any(obs.FieldErr, err))
-			<-b.asyncSlots
-			return
-		}
-
-		// Budget the whole background job (stream + webhook) by the task
-		// execution timeout so a ghosted worker can't hold a goroutine
-		// past server shutdown.
+		// Spawn-before-ack: the goroutine starts BEFORE we attempt to
+		// write the 202 body. This preserves the contract "if the client
+		// got a 202 with a task_id, the task is being executed." A failed
+		// ack write (client hung up between header and body) leaves the
+		// background task running so a webhook delivery can still fire,
+		// rather than silently dropping committed work.
 		taskCtx, cancelTask := context.WithTimeout(rootCtx, network.TaskExecutionTimeout)
-
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
@@ -132,6 +133,19 @@ func (b *Boss) asyncExecuteHandler(rootCtx context.Context, wg *sync.WaitGroup)
 
 			b.runAsyncTask(taskCtx, peerID, taskID, req)
 		}()
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusAccepted)
+		if err := json.NewEncoder(w).Encode(map[string]string{
+			"task_id": taskID,
+			"status":  "queued",
+			"message": "Task dispatched to P2P mesh.",
+		}); err != nil {
+			slog.Warn("async task ack write failed; goroutine continues",
+				slog.String(obs.FieldTaskID, taskID),
+				slog.Any(obs.FieldErr, err),
+			)
+		}
 	}
 }
 

agentfm-go/internal/boss/api_handlers.go

@@ -115,7 +115,8 @@ func (b *Boss) handleExecuteTask(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	pterm.Info.Printfln("📡 API Gateway routing task %s to Worker %s...", req.TaskID[:8], pterm.Cyan(peerID.String()[:8]))
+	pterm.Info.Printfln("📡 API Gateway routing task %s to Worker %s...",
+		shortID(req.TaskID, 8), pterm.Cyan(peerID.String()[:8]))
 
 	// Tie the dial to the inbound HTTP request's context so a client
 	// hanging up aborts the libp2p dial instead of waiting out the full

agentfm-go/internal/boss/api_test.go

@@ -281,6 +281,32 @@ func TestHandleExecuteTask_InvalidWorkerIDFormat(t *testing.T) {
 	}
 }
 
+// TestShortID_GuardsShortInput: regression for round-2 audit finding I-8.
+// The routing log line in handleExecuteTask previously sliced
+// req.TaskID[:8] directly, which panics when a user-supplied TaskID is
+// shorter than 8 chars. shortID() caps the slice safely. We unit-test
+// the helper directly because invoking the full handler path triggers
+// a pre-existing pterm-spinner concurrency issue under -race.
+func TestShortID_GuardsShortInput(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		in   string
+		n    int
+		want string
+	}{
+		{"abc", 8, "abc"},
+		{"task_abcdef12345", 8, "task_abc"},
+		{"", 8, ""},
+		{"x", 0, ""},
+		{"exactly8", 8, "exactly8"},
+	}
+	for _, tc := range cases {
+		if got := shortID(tc.in, tc.n); got != tc.want {
+			t.Errorf("shortID(%q, %d) = %q; want %q", tc.in, tc.n, got, tc.want)
+		}
+	}
+}
+
 // --- POST /api/execute/async -----------------------------------------------
 
 // TestAsyncExecuteHandler_MethodNotAllowed covers the factory-returned

agentfm-go/internal/boss/boss.go

@@ -100,7 +100,10 @@ func (b *Boss) listenTelemetry(ctx context.Context) {
 	// Periodic pruner: evicts workers whose libp2p connection has dropped.
 	// Centralised here so handleGetWorkers and the TUI tick can be pure
 	// reads (no side effects on a GET request).
-	pruneTicker := time.NewTicker(30 * time.Second)
+	// Tick faster than staleTelemetryTimeout (15s) so the lastSeen-based
+	// eviction is responsive — a 30s tick would let stale workers linger
+	// in the radar for almost half a minute past the staleness threshold.
+	pruneTicker := time.NewTicker(5 * time.Second)
 	defer pruneTicker.Stop()
 
 	msgCh := make(chan *pubsubMsg, 1)
@@ -152,9 +155,15 @@ type pubsubMsg struct {
 // pruneDisconnectedWorkers walks activeWorkers and evicts any peer the
 // libp2p host is no longer connected to. Runs under a write lock; cheap
 // because the map is bounded by mesh size (typically tens of peers).
+// staleTelemetryTimeout is the upper bound on how long a worker can go
+// without a telemetry pulse before the pruner evicts it. Mirrors the
+// previous 15s ad-hoc value that lived inline in ui.go's draw loop.
+const staleTelemetryTimeout = 15 * time.Second
+
 func (b *Boss) pruneDisconnectedWorkers() {
 	b.mu.Lock()
 	defer b.mu.Unlock()
+	now := time.Now()
 	for peerIDStr := range b.activeWorkers {
 		pID, err := peer.Decode(peerIDStr)
 		if err != nil {
@@ -165,6 +174,11 @@ func (b *Boss) pruneDisconnectedWorkers() {
 		if b.node.Host.Network().Connectedness(pID) != netcore.Connected {
 			delete(b.activeWorkers, peerIDStr)
 			delete(b.lastSeen, peerIDStr)
+			continue
+		}
+		if seen, ok := b.lastSeen[peerIDStr]; ok && now.Sub(seen) > staleTelemetryTimeout {
+			delete(b.activeWorkers, peerIDStr)
+			delete(b.lastSeen, peerIDStr)
 		}
 	}
 	metrics.WorkersOnline.Set(float64(len(b.activeWorkers)))
@@ -184,3 +198,14 @@ func (tr *timeoutReader) Read(p []byte) (n int, err error) {
 	}
 	return tr.stream.Read(p)
 }
+
+// shortID returns the first n runes of s, or s when shorter. Used for
+// log/UI snippets where a short identifier prefix is enough for humans
+// to correlate. Defends against panics on user-supplied IDs that fall
+// short of the slice length.
+func shortID(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n]
+}

agentfm-go/internal/boss/ui.go

@@ -62,24 +62,17 @@ func (b *Boss) selectWorkerInteractive(parentCtx context.Context) (types.WorkerP
 	defer area.Stop()
 
 	draw := func() {
-		b.mu.Lock()
-
-		for peerID, seen := range b.lastSeen {
-			if time.Since(seen) > 15*time.Second {
-				delete(b.activeWorkers, peerID)
-				delete(b.lastSeen, peerID)
-			}
-		}
-
+		// Pure read: pruning is handled by listenTelemetry's pruneTicker
+		// (boss.go) so the TUI and /api/workers always agree on which
+		// workers are visible. RLock is sufficient.
+		b.mu.RLock()
 		nextList := make([]types.WorkerProfile, 0, len(b.activeWorkers))
 		for _, w := range b.activeWorkers {
 			nextList = append(nextList, w)
 		}
-
 		activeCount := len(b.activeWorkers)
 		peerCount := len(b.node.PubSub.ListPeers(network.TelemetryTopic))
-
-		b.mu.Unlock()
+		b.mu.RUnlock()
 
 		sort.Slice(nextList, func(i, j int) bool {
 			return nextList[i].PeerID < nextList[j].PeerID

agentfm-go/internal/network/artifacts.go

@@ -20,7 +20,12 @@ import (
 	"github.com/pterm/pterm"
 )
 
-var safeTaskIDPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}$`)
+// SafeTaskIDPattern bounds the alphabet of a task ID that's used as a
+// path component (the artifact zip filename, the per-task /tmp/output
+// bind-mount). Exported so other packages (e.g. boss/api_async.go's
+// waitForArtifact) can apply the same defense-in-depth check rather
+// than trusting a string they got from upstream code.
+var SafeTaskIDPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}$`)
 
 func SendArtifacts(ctx context.Context, h host.Host, bossID peer.ID, zipFilePath string, taskID string) error {
 	fmt.Println("📦 Opening secure artifact channel to Boss...")
@@ -157,7 +162,7 @@ func HandleArtifactStream(stream network.Stream) {
 
 	taskID := string(idBytes)
 	safeTaskID := filepath.Base(filepath.Clean(taskID))
-	if !safeTaskIDPattern.MatchString(safeTaskID) {
+	if !SafeTaskIDPattern.MatchString(safeTaskID) {
 		safeTaskID = fmt.Sprintf("fallback_%d", time.Now().UnixNano())
 	}
 

agentfm-go/internal/network/discovery.go

@@ -103,6 +103,11 @@ func discoverPeers(ctx context.Context, h host.Host, routingDiscovery *routing.R
 				dialCtx, dialCancel := context.WithTimeout(ctx, StreamDialTimeout)
 				if err := h.Connect(dialCtx, p); err == nil {
 					fmt.Printf("\n🌍 [DHT Fallback] Successfully connected to peer: %s\n", p.ID.String()[:8])
+				} else {
+					slog.Debug("dht fallback dial failed",
+						slog.String(obs.FieldPeerID, p.ID.String()),
+						slog.Any(obs.FieldErr, err),
+					)
 				}
 				dialCancel()
 			}

agentfm-go/internal/network/host.go

@@ -17,18 +17,18 @@ import (
 	"github.com/libp2p/go-libp2p/p2p/host/autonat"
 )
 
-// loadOrGenerateIdentity returns the persistent Ed25519 private key for
-// this node, creating one on first boot. Stable peer IDs across restarts
-// are important: they let other nodes cache our address and skip the DHT
-// lookup next time, and private swarms often allowlist peers by ID.
-func loadOrGenerateIdentity(mode string) (crypto.PrivKey, error) {
-	keyPath := fmt.Sprintf(".agentfm_%s_identity.key", mode)
-
-	// Try the existing identity first. A read failure is fine on the
-	// very first boot, but a successful read that fails to unmarshal
-	// means the file on disk is corrupt. We surface that so the
-	// operator realises their peer ID is about to change, instead of
-	// silently regenerating.
+// LoadOrGenerateIdentity returns the persistent Ed25519 private key at
+// keyPath, creating one on first boot. Shared by both binaries:
+// internal/network's mode-suffixed boss/worker/api identities, and the
+// dedicated relay binary's relay_identity.key. Stable peer IDs across
+// restarts let other nodes cache our address and skip DHT lookups, and
+// private swarms often allowlist peers by ID.
+//
+// A read failure is fine on the very first boot, but a successful read
+// that fails to unmarshal means the file on disk is corrupt. We surface
+// that so the operator realises their peer ID is about to change instead
+// of silently regenerating.
+func LoadOrGenerateIdentity(keyPath string) (crypto.PrivKey, error) {
 	if keyBytes, err := os.ReadFile(keyPath); err == nil {
 		priv, err := crypto.UnmarshalPrivateKey(keyBytes)
 		if err == nil {
@@ -62,6 +62,13 @@ func loadOrGenerateIdentity(mode string) (crypto.PrivKey, error) {
 	return priv, nil
 }
 
+// loadOrGenerateIdentity preserves the legacy package-internal call sites
+// (createHost) that pass a mode string and want the .agentfm_<mode>_identity.key
+// naming convention. New callers should use LoadOrGenerateIdentity directly.
+func loadOrGenerateIdentity(mode string) (crypto.PrivKey, error) {
+	return LoadOrGenerateIdentity(fmt.Sprintf(".agentfm_%s_identity.key", mode))
+}
+
 // createHost assembles the libp2p Host with the correct options for this
 // role: PSK for private swarms, circuit relay in the right direction,
 // NAT port mapping, and the AutoNAT reachability probe for non-relay
@@ -116,7 +123,9 @@ func createHost(cfg Config, bootstrapAddr string) (host.Host, error) {
 
 	if cfg.Mode != "relay" {
 		if _, err = autonat.New(h); err != nil {
-			fmt.Printf("⚠️  [NAT] Failed to start AutoNAT service: %v\n", err)
+			slog.Warn("autonat unavailable; reachability autodetection disabled",
+				slog.Any(obs.FieldErr, err),
+			)
 		} else {
 			fmt.Println("🌐 [NAT] AutoNAT service started. Testing public reachability...")
 		}

agentfm-go/internal/network/p2p.go

@@ -3,6 +3,9 @@ package network
 import (
 	"context"
 	"fmt"
+	"log/slog"
+
+	"agentfm/internal/obs"
 
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -56,7 +59,13 @@ func Setup(ctx context.Context, cfg Config) (*MeshNode, error) {
 
 	var relayPeerID peer.ID
 	if cfg.Mode != "relay" && bootstrapAddr != "" {
-		if relayInfo, err := parseRelayInfo(bootstrapAddr); err == nil {
+		relayInfo, err := parseRelayInfo(bootstrapAddr)
+		if err != nil {
+			slog.Warn("invalid bootstrap multiaddr; skipping lighthouse",
+				slog.String("bootstrap", bootstrapAddr),
+				slog.Any(obs.FieldErr, err),
+			)
+		} else {
 			relayPeerID = relayInfo.ID
 			connectToLighthouse(ctx, h, relayInfo)
 		}