Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 08f3db318f | |||
| 69e2800ef3 | |||
| a8fcb404be |
@@ -81,6 +81,11 @@ var querySources = []querySource{
|
||||
// wording, so "какая температура на улице?" still reaches the weather
|
||||
// source.
|
||||
{"home", (*reactiveHandler).queryHome},
|
||||
// Next to "home" and for the same reason: "какие устройства в сети?" is a
|
||||
// question about the LAN, and the recall pass would otherwise answer it
|
||||
// from an old note about the router. Its matcher needs a network word plus
|
||||
// an ask plus a device noun, so "интернет не работает" is untouched.
|
||||
{"network", (*reactiveHandler).queryNetwork},
|
||||
{"calendar", (*reactiveHandler).queryCalendar},
|
||||
{"weather", (*reactiveHandler).queryWeather},
|
||||
{"embed", (*reactiveHandler).queryEmbed},
|
||||
@@ -299,6 +304,22 @@ func (h *reactiveHandler) queryHome(ctx context.Context, t *queryTurn) (string,
|
||||
return h.home.homeSummary(ctxH)
|
||||
}
|
||||
|
||||
// queryNetwork answers a question about the LAN with a bounded scan. There is
|
||||
// no confirm turn because nothing is changed, and no way to widen the range
|
||||
// because Scan takes no target — the utterance selects the question, never the
|
||||
// subnet.
|
||||
func (h *reactiveHandler) queryNetwork(ctx context.Context, t *queryTurn) (string, bool) {
|
||||
if !isNetworkQuery(t.dec.Utterance) {
|
||||
return "", false
|
||||
}
|
||||
if h.netscan == nil {
|
||||
// Claim the turn: "сканирование не настроено" is true, and general
|
||||
// knowledge would answer with an invented list of devices.
|
||||
return "сканирование сети не настроено.", true
|
||||
}
|
||||
return h.netscan.scanSummary(ctx)
|
||||
}
|
||||
|
||||
func (h *reactiveHandler) queryWeather(ctx context.Context, t *queryTurn) (string, bool) {
|
||||
if !isWeatherQuery(t.dec.Utterance) {
|
||||
return "", false
|
||||
|
||||
@@ -72,6 +72,7 @@ var praxisCapabilities = []praxisCapability{
|
||||
},
|
||||
},
|
||||
listChangesCapability{},
|
||||
entityAttentionCapability{},
|
||||
}
|
||||
|
||||
// handlePraxisAct — dispatches ecosystem tool acts through the Praxis tools API.
|
||||
@@ -190,6 +191,106 @@ func (listChangesCapability) handle(ctx context.Context, h *reactiveHandler, px
|
||||
return "изменения: " + strings.Join(parts, "; ")
|
||||
}
|
||||
|
||||
// entityAttentionCapability answers "what's going on with X" by resolving X to
|
||||
// a canonical Nexus entity and asking Praxis for that entity's attention items
|
||||
// (Vikunja #272). Unlike listAttentionCapability it is scoped: the entity_id
|
||||
// travels to Praxis as a query parameter instead of Maven filtering an unscoped
|
||||
// list client-side, which is what makes the ref canonical end to end.
|
||||
//
|
||||
// It also folds in what Maven herself knows about the same entity — facts the
|
||||
// enrichment worker has already resolved to that entity_id — so one question
|
||||
// gets one answer across both stores.
|
||||
type entityAttentionCapability struct{}
|
||||
|
||||
func (entityAttentionCapability) aliases() []string {
|
||||
return []string{"entity_attention", "что с", "как дела у", "статус"}
|
||||
}
|
||||
|
||||
func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler, px *praxisClient, dec router.Decision) string {
|
||||
subject := dec.Slots.Value
|
||||
if subject == "" {
|
||||
subject = dec.Slots.Text
|
||||
}
|
||||
if subject == "" {
|
||||
return "про что именно спросить?"
|
||||
}
|
||||
if h.ecosystem == nil || h.ecosystem.nexus == nil {
|
||||
// Without Nexus there is no canonical ref to scope by. Say so rather
|
||||
// than quietly answering about something else.
|
||||
return "не могу связать это с сущностью — Nexus не настроен."
|
||||
}
|
||||
|
||||
entityID, displayName, ambiguous, err := h.ecosystem.resolveEntityReference(ctx, subject, nil)
|
||||
if err != nil {
|
||||
log.Printf("ecosystem: entity attention resolve %q: %v", subject, err)
|
||||
return "экосистема недоступна, попробуй ещё раз."
|
||||
}
|
||||
if len(ambiguous) > 0 {
|
||||
return "уточни, что именно: " + strings.Join(ambiguous, ", ") + "?"
|
||||
}
|
||||
if entityID == "" {
|
||||
return "не знаю такой сущности."
|
||||
}
|
||||
if displayName == "" {
|
||||
displayName = subject
|
||||
}
|
||||
|
||||
items, err := px.ListAttentionForEntity(ctx, entityID, 20)
|
||||
if err != nil {
|
||||
log.Printf("ecosystem: praxis attention for %s: %v", entityID, err)
|
||||
return "не могу сейчас узнать, что требует внимания по «" + displayName + "»."
|
||||
}
|
||||
h.recordPraxisTrace(ctx, "entity_attention", map[string]any{
|
||||
"entity_id": entityID, "count": len(items),
|
||||
})
|
||||
|
||||
var parts []string
|
||||
for _, item := range items {
|
||||
title, _ := item["title"].(string)
|
||||
if title == "" {
|
||||
continue
|
||||
}
|
||||
parts = append(parts, title)
|
||||
// Same surfaced != acknowledged rule as the unscoped digest.
|
||||
if id, ok := item["id"].(string); ok && id != "" {
|
||||
if _, err := px.Surface(ctx, id); err != nil {
|
||||
log.Printf("ecosystem: praxis surface %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if known := h.localFactsForEntity(ctx, entityID); known != "" {
|
||||
parts = append(parts, known)
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return "по «" + displayName + "» ничего нет."
|
||||
}
|
||||
return "по «" + displayName + "»: " + strings.Join(parts, "; ")
|
||||
}
|
||||
|
||||
// localFactsForEntity summarises Maven's own facts already resolved to this
|
||||
// canonical entity. Empty when the store is unavailable or nothing matched —
|
||||
// entity-scoped memory is an enrichment of the answer, never a precondition.
|
||||
func (h *reactiveHandler) localFactsForEntity(ctx context.Context, entityID string) string {
|
||||
if h.dataStore == nil || entityID == "" {
|
||||
return ""
|
||||
}
|
||||
facts, err := h.dataStore.FactsByEntity(ctx, entityID, 3)
|
||||
if err != nil {
|
||||
log.Printf("ecosystem: facts by entity %s: %v", entityID, err)
|
||||
return ""
|
||||
}
|
||||
var parts []string
|
||||
for _, f := range facts {
|
||||
if f.Value != "" {
|
||||
parts = append(parts, f.Value)
|
||||
}
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return ""
|
||||
}
|
||||
return "я помню: " + strings.Join(parts, ", ")
|
||||
}
|
||||
|
||||
// recordPraxisTrace — writes a fact recording a cross-service ecosystem call.
|
||||
// The fact is stored with source "praxis:trace" so the proactive loop can
|
||||
// reference it and the dashboard can display recent ecosystem activity.
|
||||
|
||||
@@ -0,0 +1,312 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
hexisclient "github.com/kami/hexis/pkg/client"
|
||||
|
||||
"github.com/kami/maven/internal/ipc"
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
// Phase-5 hardening suite (Vikunja #276). Everything here drives the shared
|
||||
// fake ecosystem (fakeecosystem_test.go) rather than one-off inline handlers,
|
||||
// so the same fault levers — SetFault, SetBody, SetDelay — cover every
|
||||
// service. What is asserted is the degraded-mode contract:
|
||||
//
|
||||
// - services degrade independently: one outage never mutes the others,
|
||||
// - a degraded reply is never silent, never fabricated, never "success",
|
||||
// - contract drift (old shape, unknown fields, garbage) is survivable,
|
||||
// - Maven never acts on an ambiguous target and never chains
|
||||
// Praxis observation into Hexis execution on its own.
|
||||
|
||||
// ecoHandler wires a handler against whichever of the three fakes is given
|
||||
// (pass nil to leave a service unconfigured, which is a different state from
|
||||
// "configured but down").
|
||||
func ecoHandler(t *testing.T, nexus, praxis, hexis *fakeServer) *reactiveHandler {
|
||||
t.Helper()
|
||||
st := newTestStore(t)
|
||||
clock := newFakeClock(time.Date(2026, 8, 1, 9, 0, 0, 0, time.UTC))
|
||||
w := &ecosystemWiring{}
|
||||
if nexus != nil {
|
||||
w.nexus = newNexusClient(nexus.URL)
|
||||
}
|
||||
if praxis != nil {
|
||||
w.praxis = newPraxisClient(praxis.URL)
|
||||
}
|
||||
if hexis != nil {
|
||||
w.hexis = hexisclient.New(hexis.URL)
|
||||
}
|
||||
return &reactiveHandler{
|
||||
api: ipc.NewStoreAPI(st),
|
||||
dataStore: st,
|
||||
now: clock.Now,
|
||||
ecosystem: w,
|
||||
}
|
||||
}
|
||||
|
||||
func traceFacts(t *testing.T, h *reactiveHandler) []store.Fact {
|
||||
t.Helper()
|
||||
facts, err := h.dataStore.RecentFacts(context.Background(), 50)
|
||||
if err != nil {
|
||||
t.Fatalf("read facts: %v", err)
|
||||
}
|
||||
var out []store.Fact
|
||||
for _, f := range facts {
|
||||
if f.Source == "praxis:trace" {
|
||||
out = append(out, f)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func restartCaps() string {
|
||||
return fixtureHexisCapabilities(map[string]any{
|
||||
"id": "cap_restart", "name": "restart", "read_only": true,
|
||||
})
|
||||
}
|
||||
|
||||
// TestEcosystem_OutagesAreIndependent: Praxis being down must not disable the
|
||||
// Nexus+Hexis action path, and vice versa. A shared "ecosystem is broken"
|
||||
// mode would take away working capability for no reason.
|
||||
func TestEcosystem_OutagesAreIndependent(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(map[string]any{
|
||||
"id": "item_1", "title": "disk almost full", "importance": 3.0,
|
||||
}))
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, praxis, hexis)
|
||||
|
||||
praxis.SetFault(503)
|
||||
if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); !strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("praxis outage must not block the hexis path, got %q", reply)
|
||||
}
|
||||
|
||||
praxis.SetFault(0)
|
||||
hexis.SetFault(503)
|
||||
nexus.SetFault(503)
|
||||
reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
|
||||
if !strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("nexus/hexis outage must not block the praxis digest, got %q", reply)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_MalformedNexusResponseFailsClosed: a 200 carrying garbage is a
|
||||
// dependency failure, not "no such entity". It must stop before Hexis.
|
||||
func TestEcosystem_MalformedNexusResponseFailsClosed(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
|
||||
nexus.SetBody(`{"status":"resolved","entity":`)
|
||||
reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
|
||||
if reply == "" || strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("malformed nexus body must degrade, got %q", reply)
|
||||
}
|
||||
if hexis.Count("", "/api/v1") != 0 {
|
||||
t.Fatal("hexis must not be contacted after a malformed nexus response")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_UnknownContractFieldsTolerated: a newer Nexus adding fields
|
||||
// must not break an older Maven. Same for the older flat resolve shape.
|
||||
func TestEcosystem_UnknownContractFieldsTolerated(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
for name, body := range map[string]string{
|
||||
"future": fixtureNexusResolvedFuture("ent_muzick", "Muzick indexer", "service"),
|
||||
"flat": fixtureNexusResolvedFlat("ent_muzick", "Muzick indexer", "service"),
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
nexus := newFakeNexus(t, body)
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); !strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("%s contract shape must still resolve and execute, got %q", name, reply)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_CancelledContextDegrades: a caller hanging up (turn abandoned,
|
||||
// deadline hit) must surface as degradation, never as a fabricated result.
|
||||
func TestEcosystem_CancelledContextDegrades(t *testing.T) {
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
nexus.SetDelay(2 * time.Second)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
||||
defer cancel()
|
||||
reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
|
||||
if reply == "" || strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("cancelled resolve must degrade, got %q", reply)
|
||||
}
|
||||
if hexis.Count("", "/api/v1") != 0 {
|
||||
t.Fatal("hexis must not be contacted after a cancelled resolve")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_ExecutionFailureIsNotSuccess: Hexis answering 200 with
|
||||
// status=failed is a partial failure — the call worked, the command did not.
|
||||
// Maven must report it as a failure and must not write a success trace.
|
||||
func TestEcosystem_ExecutionFailureIsNotSuccess(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecutionFailed("exec_1", "unit not found"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
|
||||
reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
|
||||
if strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("failed execution must not read as success, got %q", reply)
|
||||
}
|
||||
if reply == "" {
|
||||
t.Fatal("failed execution must say something")
|
||||
}
|
||||
for _, f := range traceFacts(t, h) {
|
||||
if strings.HasPrefix(f.Key, "praxis:hexis:") {
|
||||
t.Fatalf("failed execution must not write a success trace: %+v", f)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_AmbiguousTargetBlocksExecution: ambiguity blocks mutation, and
|
||||
// the clarification must name the candidates rather than pick one.
|
||||
func TestEcosystem_AmbiguousTargetBlocksExecution(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusAmbiguous(
|
||||
map[string]string{"entity_id": "ent_a", "display_name": "Muzick indexer"},
|
||||
map[string]string{"entity_id": "ent_b", "display_name": "Muzick web"},
|
||||
))
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
|
||||
reply := h.handleHexisAct(ctx, actDec("muzick"))
|
||||
if !strings.Contains(reply, "Muzick indexer") || !strings.Contains(reply, "Muzick web") {
|
||||
t.Fatalf("ambiguous resolve must list candidates, got %q", reply)
|
||||
}
|
||||
if hexis.Count("POST", "/api/v1/execute") != 0 {
|
||||
t.Fatal("ambiguous target must never execute")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_NoAutonomousPraxisToHexis: reading the attention digest is an
|
||||
// observation. Maven must never turn an observed problem into a Hexis command
|
||||
// by herself — she is not autonomous.
|
||||
func TestEcosystem_NoAutonomousPraxisToHexis(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(
|
||||
map[string]any{"id": "item_1", "title": "muzick indexer is down", "importance": 4.0, "rule": "service_down"},
|
||||
))
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, praxis, hexis)
|
||||
|
||||
_ = h.handlePraxisAct(ctx, praxisActDec("list_attention"))
|
||||
if hexis.Count("", "/api/v1") != 0 {
|
||||
t.Fatal("attention digest must not contact hexis on its own")
|
||||
}
|
||||
if nexus.Count("", "/api/v1/resolve") != 0 {
|
||||
t.Fatal("attention digest must not resolve targets for autonomous action")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_MutatingCapabilityWaitsForConfirmation: a non-read-only
|
||||
// capability parks for an explicit spoken confirm bound to capability+target.
|
||||
func TestEcosystem_MutatingCapabilityWaitsForConfirmation(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
caps := fixtureHexisCapabilities(map[string]any{"id": "cap_restart", "name": "restart", "read_only": false})
|
||||
hexis := newFakeHexis(t, caps, fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
|
||||
reply := h.handleHexisAct(ctx, actDec("restart"))
|
||||
if !strings.Contains(reply, "restart") || !strings.Contains(reply, "да") {
|
||||
t.Fatalf("mutating capability must ask for confirmation, got %q", reply)
|
||||
}
|
||||
if hexis.Count("POST", "/api/v1/execute") != 0 {
|
||||
t.Fatal("mutating capability must not execute before confirmation")
|
||||
}
|
||||
h.mu.Lock()
|
||||
pending := h.pendingHexis
|
||||
h.mu.Unlock()
|
||||
if pending == nil || pending.capabilityID != "cap_restart" || pending.entityID != "ent_muzick" {
|
||||
t.Fatalf("confirmation must be bound to capability+target, got %+v", pending)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_SurfaceFailureStillDelivers: surfacing is bookkeeping. If the
|
||||
// surface call fails the digest must still be spoken — a partial failure
|
||||
// downgrades bookkeeping, not the answer.
|
||||
func TestEcosystem_SurfaceFailureStillDelivers(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
praxis := newFakeServer(t, map[string]http.HandlerFunc{
|
||||
"GET /api/v1/tools/attention": jsonHandler(200, fixturePraxisAttentionItems(
|
||||
map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
|
||||
)),
|
||||
"POST /api/v1/tools/surface": jsonHandler(500, `{"error":"boom"}`),
|
||||
})
|
||||
h := ecoHandler(t, nil, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
|
||||
if !strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("failed surface must not swallow the digest, got %q", reply)
|
||||
}
|
||||
if praxis.Count("POST", "/api/v1/tools/surface") == 0 {
|
||||
t.Fatal("expected the surface attempt")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_TotalOutageSaysSoForEveryPath: with all three down, every
|
||||
// entry point degrades explicitly instead of returning empty or inventing.
|
||||
func TestEcosystem_TotalOutageSaysSoForEveryPath(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems())
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
for _, fs := range []*fakeServer{nexus, praxis, hexis} {
|
||||
fs.SetFault(503)
|
||||
}
|
||||
h := ecoHandler(t, nexus, praxis, hexis)
|
||||
|
||||
for name, reply := range map[string]string{
|
||||
"hexis act": h.handleHexisAct(ctx, actDec("muzick indexer")),
|
||||
"attention": h.handlePraxisAct(ctx, praxisActDec("list_attention")),
|
||||
"changes": h.handlePraxisAct(ctx, praxisActDec("list_changes")),
|
||||
"acknowledge": h.handlePraxisAct(ctx, praxisActDec("acknowledge_item")),
|
||||
} {
|
||||
if reply == "" {
|
||||
t.Errorf("%s: total outage must not answer with silence", name)
|
||||
}
|
||||
if strings.Contains(reply, "выполнена") {
|
||||
t.Errorf("%s: total outage must not claim success: %q", name, reply)
|
||||
}
|
||||
}
|
||||
if len(traceFacts(t, h)) != 0 {
|
||||
t.Fatal("a total outage must not leave success traces behind")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_RecoveryAfterOutageNeedsNoRestart: once the dependency comes
|
||||
// back the very next turn works — no cached failure state, no restart.
|
||||
func TestEcosystem_RecoveryAfterOutageNeedsNoRestart(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(
|
||||
map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
|
||||
))
|
||||
h := ecoHandler(t, nil, praxis, nil)
|
||||
|
||||
praxis.SetFault(503)
|
||||
if reply := h.handlePraxisAct(ctx, praxisActDec("list_attention")); strings.Contains(reply, "disk") {
|
||||
t.Fatalf("outage must not serve content, got %q", reply)
|
||||
}
|
||||
praxis.SetFault(0)
|
||||
if reply := h.handlePraxisAct(ctx, praxisActDec("list_attention")); !strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("recovery must work on the next turn, got %q", reply)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/router"
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
// Entity-ref propagation, Maven side (Vikunja #272): the canonical Nexus
|
||||
// entity_id must reach Praxis as a query scope rather than being resolved and
|
||||
// then thrown away, and the enrichment that produces those ids must degrade
|
||||
// visibly instead of silently.
|
||||
|
||||
func entityAttentionDec(subject string) router.Decision {
|
||||
return router.Decision{
|
||||
Intent: router.IntentAct,
|
||||
Slots: router.Slots{Fn: "entity_attention", HasFn: true, Value: subject},
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_ScopesPraxisByCanonicalID: the resolved id must travel
|
||||
// to Praxis in the request, not be used for client-side filtering.
|
||||
func TestEntityAttention_ScopesPraxisByCanonicalID(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(
|
||||
map[string]any{"id": "item_1", "title": "indexer queue is backing up", "importance": 3.0},
|
||||
))
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
|
||||
if !strings.Contains(reply, "indexer queue is backing up") {
|
||||
t.Fatalf("expected the scoped item in the reply, got %q", reply)
|
||||
}
|
||||
|
||||
var scoped bool
|
||||
for _, r := range praxis.Requests() {
|
||||
if r.Method == "GET" && strings.HasPrefix(r.Path, "/api/v1/tools/attention") &&
|
||||
strings.Contains(r.Query, "entity_id=ent_muzick") {
|
||||
scoped = true
|
||||
}
|
||||
}
|
||||
if !scoped {
|
||||
t.Fatalf("expected attention scoped by entity_id, got requests %+v", praxis.Requests())
|
||||
}
|
||||
if praxis.Count("POST", "/api/v1/tools/surface") == 0 {
|
||||
t.Error("a spoken scoped item must be surfaced, like the unscoped digest")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_FoldsInLocalFactsForSameEntity: facts the enrichment
|
||||
// worker already tagged with the same canonical id join the same answer.
|
||||
func TestEntityAttention_FoldsInLocalFactsForSameEntity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems())
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
|
||||
id, err := h.dataStore.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv,
|
||||
"descaled", "the espresso machine", "descaled in june", "infer:pref", 0.8, sql.NullInt64{})
|
||||
if err != nil {
|
||||
t.Fatalf("WriteFactAboutSubject: %v", err)
|
||||
}
|
||||
if err := h.dataStore.ResolveFactEntity(ctx, id, "ent_espresso", store.ResolutionResolved); err != nil {
|
||||
t.Fatalf("ResolveFactEntity: %v", err)
|
||||
}
|
||||
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("the espresso machine"))
|
||||
if !strings.Contains(reply, "descaled in june") {
|
||||
t.Fatalf("expected entity-scoped local facts in the reply, got %q", reply)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_AmbiguousAsksInsteadOfGuessing.
|
||||
func TestEntityAttention_AmbiguousAsksInsteadOfGuessing(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusAmbiguous(
|
||||
map[string]string{"entity_id": "ent_a", "display_name": "Muzick indexer"},
|
||||
map[string]string{"entity_id": "ent_b", "display_name": "Muzick web"},
|
||||
))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems())
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick"))
|
||||
if !strings.Contains(reply, "Muzick indexer") || !strings.Contains(reply, "Muzick web") {
|
||||
t.Fatalf("ambiguous subject must ask, got %q", reply)
|
||||
}
|
||||
if praxis.Count("GET", "/api/v1/tools/attention") != 0 {
|
||||
t.Fatal("an ambiguous subject must not be queried against praxis")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_MissingAndDegradedAreDistinct: "no such entity" and
|
||||
// "Nexus is down" must not produce the same answer.
|
||||
func TestEntityAttention_MissingAndDegradedAreDistinct(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusNotFound())
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems())
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
|
||||
missing := h.handlePraxisAct(ctx, entityAttentionDec("нечто"))
|
||||
if missing == "" {
|
||||
t.Fatal("an unknown entity must still get an answer")
|
||||
}
|
||||
|
||||
nexus.SetFault(503)
|
||||
degraded := h.handlePraxisAct(ctx, entityAttentionDec("нечто"))
|
||||
if degraded == missing {
|
||||
t.Fatalf("outage and unknown-entity must not read the same: %q", degraded)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_DelayedNexusDegradesNotHangs: a slow Nexus past the
|
||||
// caller's deadline degrades and never queries Praxis with an empty scope.
|
||||
func TestEntityAttention_DelayedNexusDegradesNotHangs(t *testing.T) {
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems())
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
nexus.SetDelay(2 * time.Second)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
||||
defer cancel()
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
|
||||
if reply == "" {
|
||||
t.Fatal("a delayed resolve must still answer")
|
||||
}
|
||||
if praxis.Count("GET", "/api/v1/tools/attention") != 0 {
|
||||
t.Fatal("praxis must not be queried without a resolved scope")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_WithoutNexusSaysSo: no Nexus means no canonical ref, so
|
||||
// the scoped query is refused rather than answered about something else.
|
||||
func TestEntityAttention_WithoutNexusSaysSo(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(
|
||||
map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
|
||||
))
|
||||
h := ecoHandler(t, nil, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
|
||||
if strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("unscoped items must not be passed off as entity-scoped, got %q", reply)
|
||||
}
|
||||
if praxis.Count("GET", "/api/v1/tools/attention") != 0 {
|
||||
t.Fatal("no canonical ref means no scoped query at all")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEnrichmentBackoff_HoldsAndReleases: repeated Nexus failures back the
|
||||
// fact off instead of hammering, and the fact is retried once the window
|
||||
// elapses. Nothing is ever given up on.
|
||||
func TestEnrichmentBackoff_HoldsAndReleases(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device"))
|
||||
st := newTestStore(t)
|
||||
if _, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes",
|
||||
"the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{}); err != nil {
|
||||
t.Fatalf("WriteFactAboutSubject: %v", err)
|
||||
}
|
||||
|
||||
clock := newFakeClock(time.Date(2026, 8, 1, 3, 0, 0, 0, time.UTC))
|
||||
w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour)
|
||||
w.now = clock.Now
|
||||
|
||||
nexus.SetFault(503)
|
||||
w.tick(ctx)
|
||||
failedCalls := nexus.Count("POST", "/api/v1/resolve")
|
||||
if failedCalls != 1 {
|
||||
t.Fatalf("expected one resolve attempt, got %d", failedCalls)
|
||||
}
|
||||
|
||||
// Immediately after a failure the fact is in backoff: no second call.
|
||||
w.tick(ctx)
|
||||
if nexus.Count("POST", "/api/v1/resolve") != failedCalls {
|
||||
t.Fatal("a fact in backoff must not be retried on the very next tick")
|
||||
}
|
||||
if s := w.status(ctx); s.Pending != 1 || s.InBackoff != 1 || s.MaxAttempts != 1 {
|
||||
t.Fatalf("degradation must be reported, got %+v", s)
|
||||
}
|
||||
|
||||
// Once the window elapses and Nexus recovers, the fact resolves.
|
||||
clock.Advance(2 * time.Minute)
|
||||
nexus.SetFault(0)
|
||||
w.tick(ctx)
|
||||
facts, err := st.FactsByEntity(ctx, "ent_espresso", 10)
|
||||
if err != nil {
|
||||
t.Fatalf("FactsByEntity: %v", err)
|
||||
}
|
||||
if len(facts) != 1 {
|
||||
t.Fatalf("expected the fact resolved after recovery, got %+v", facts)
|
||||
}
|
||||
if s := w.status(ctx); s.Pending != 0 || s.MaxAttempts != 0 {
|
||||
t.Fatalf("recovery must clear the degradation report, got %+v", s)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnrichmentBackoff_GrowsAndIsCapped(t *testing.T) {
|
||||
if enrichmentBackoff(1) != time.Minute {
|
||||
t.Fatalf("first retry should be a minute, got %v", enrichmentBackoff(1))
|
||||
}
|
||||
if enrichmentBackoff(3) != 4*time.Minute {
|
||||
t.Fatalf("third retry should be four minutes, got %v", enrichmentBackoff(3))
|
||||
}
|
||||
if enrichmentBackoff(50) != time.Hour {
|
||||
t.Fatalf("backoff must cap at an hour, got %v", enrichmentBackoff(50))
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/store"
|
||||
@@ -24,10 +25,69 @@ type factEnrichmentWorker struct {
|
||||
eco *ecosystemWiring
|
||||
interval time.Duration
|
||||
batch int // facts resolved per tick; keeps a single slow tick bounded
|
||||
now func() time.Time
|
||||
|
||||
// Retry state for facts whose resolution failed transiently. Kept in
|
||||
// memory rather than in the DB: a restart legitimately retries
|
||||
// everything, and the backoff exists to spare a struggling Nexus, not
|
||||
// to be durable. A fact is never given up on — degraded means slower,
|
||||
// not dropped.
|
||||
mu sync.Mutex
|
||||
attempt map[int64]int // fact id → consecutive failures
|
||||
nextTry map[int64]time.Time // fact id → earliest retry
|
||||
skipped int // facts held back by backoff on the last tick
|
||||
}
|
||||
|
||||
// enrichmentBackoff is the wait before retrying a fact after n consecutive
|
||||
// failures, capped so a long Nexus outage still retries about hourly.
|
||||
func enrichmentBackoff(n int) time.Duration {
|
||||
d := time.Minute
|
||||
for i := 1; i < n && d < time.Hour; i++ {
|
||||
d *= 2
|
||||
}
|
||||
if d > time.Hour {
|
||||
d = time.Hour
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval time.Duration) *factEnrichmentWorker {
|
||||
return &factEnrichmentWorker{store: st, eco: eco, interval: interval, batch: 20}
|
||||
return &factEnrichmentWorker{
|
||||
store: st,
|
||||
eco: eco,
|
||||
interval: interval,
|
||||
batch: 20,
|
||||
now: time.Now,
|
||||
attempt: map[int64]int{},
|
||||
nextTry: map[int64]time.Time{},
|
||||
}
|
||||
}
|
||||
|
||||
// enrichmentStatus is what the worker reports about its own health: how many
|
||||
// facts are waiting, how many are currently in backoff, and the worst retry
|
||||
// count seen. Degradation is reported, never hidden — a Nexus that has been
|
||||
// down all day must be visible as a backlog, not as facts that silently
|
||||
// never got tagged.
|
||||
type enrichmentStatus struct {
|
||||
Pending int
|
||||
InBackoff int
|
||||
MaxAttempts int
|
||||
}
|
||||
|
||||
func (w *factEnrichmentWorker) status(ctx context.Context) enrichmentStatus {
|
||||
var st enrichmentStatus
|
||||
if pending, err := w.store.PendingFactResolutions(ctx, 1000); err == nil {
|
||||
st.Pending = len(pending)
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
st.InBackoff = w.skipped
|
||||
for _, n := range w.attempt {
|
||||
if n > st.MaxAttempts {
|
||||
st.MaxAttempts = n
|
||||
}
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
func (w *factEnrichmentWorker) run(ctx context.Context) {
|
||||
@@ -57,18 +117,50 @@ func (w *factEnrichmentWorker) tick(ctx context.Context) {
|
||||
log.Printf("factenrichment: list pending: %v", err)
|
||||
return
|
||||
}
|
||||
skipped, failed := 0, 0
|
||||
for _, f := range pending {
|
||||
w.resolveOne(ctx, f)
|
||||
if !w.due(f.ID) {
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
if !w.resolveOne(ctx, f) {
|
||||
failed++
|
||||
}
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.skipped = skipped
|
||||
w.mu.Unlock()
|
||||
if failed > 0 {
|
||||
log.Printf("factenrichment: %d/%d resolutions failed this tick, %d held in backoff",
|
||||
failed, len(pending), skipped)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) {
|
||||
// due reports whether a fact's backoff window has elapsed.
|
||||
func (w *factEnrichmentWorker) due(id int64) bool {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
next, ok := w.nextTry[id]
|
||||
return !ok || !w.now().Before(next)
|
||||
}
|
||||
|
||||
// resolveOne resolves one pending fact. It returns false when the attempt
|
||||
// failed transiently: the fact stays pending and is retried on a backoff.
|
||||
func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) bool {
|
||||
entityID, _, ambiguous, err := w.eco.resolveEntityReference(ctx, f.Subject, nil)
|
||||
if err != nil {
|
||||
// Transient (Nexus unreachable) — leave pending, retry next tick.
|
||||
// Transient (Nexus unreachable) — leave pending, back off, retry later.
|
||||
log.Printf("factenrichment: resolve fact %d subject %q: %v", f.ID, f.Subject, err)
|
||||
return
|
||||
w.mu.Lock()
|
||||
w.attempt[f.ID]++
|
||||
w.nextTry[f.ID] = w.now().Add(enrichmentBackoff(w.attempt[f.ID]))
|
||||
w.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
w.mu.Lock()
|
||||
delete(w.attempt, f.ID)
|
||||
delete(w.nextTry, f.ID)
|
||||
w.mu.Unlock()
|
||||
state := store.ResolutionNotFound
|
||||
switch {
|
||||
case entityID != "":
|
||||
@@ -78,5 +170,7 @@ func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) {
|
||||
}
|
||||
if err := w.store.ResolveFactEntity(ctx, f.ID, entityID, state); err != nil {
|
||||
log.Printf("factenrichment: record resolution for fact %d: %v", f.ID, err)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -14,7 +14,9 @@ import (
|
||||
type capturedRequest struct {
|
||||
Method string
|
||||
Path string
|
||||
Query string
|
||||
Body []byte
|
||||
Header http.Header
|
||||
}
|
||||
|
||||
// fakeServer is the common shell behind fakeNexus/fakePraxis/fakeHexis: an
|
||||
@@ -27,7 +29,9 @@ type fakeServer struct {
|
||||
|
||||
mu sync.Mutex
|
||||
requests []capturedRequest
|
||||
fault int // non-zero: every request gets this HTTP status instead of routing
|
||||
fault int // non-zero: every request gets this HTTP status instead of routing
|
||||
garbage string // non-empty: returned 200 verbatim instead of routing (malformed-contract lever)
|
||||
delay time.Duration
|
||||
}
|
||||
|
||||
// newFakeServer starts a server dispatching to routes keyed by "METHOD
|
||||
@@ -47,14 +51,34 @@ func newFakeServer(t *testing.T, routes map[string]http.HandlerFunc) *fakeServer
|
||||
}
|
||||
}
|
||||
fs.mu.Lock()
|
||||
fs.requests = append(fs.requests, capturedRequest{Method: r.Method, Path: r.URL.Path, Body: body})
|
||||
fs.requests = append(fs.requests, capturedRequest{
|
||||
Method: r.Method,
|
||||
Path: r.URL.Path,
|
||||
Query: r.URL.RawQuery,
|
||||
Body: body,
|
||||
Header: r.Header.Clone(),
|
||||
})
|
||||
fault := fs.fault
|
||||
garbage := fs.garbage
|
||||
delay := fs.delay
|
||||
fs.mu.Unlock()
|
||||
|
||||
if delay > 0 {
|
||||
select {
|
||||
case <-time.After(delay):
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
if fault != 0 {
|
||||
http.Error(w, "injected fault", fault)
|
||||
return
|
||||
}
|
||||
if garbage != "" {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Write([]byte(garbage))
|
||||
return
|
||||
}
|
||||
|
||||
for key, handler := range routes {
|
||||
method, prefix := splitRouteKey(key)
|
||||
@@ -90,6 +114,36 @@ func (fs *fakeServer) SetFault(status int) {
|
||||
fs.fault = status
|
||||
}
|
||||
|
||||
// SetBody makes every subsequent request answer 200 with the given body,
|
||||
// bypassing the route table. Used to serve a malformed or contract-violating
|
||||
// payload where the transport itself is healthy. Pass "" to clear it.
|
||||
func (fs *fakeServer) SetBody(body string) {
|
||||
fs.mu.Lock()
|
||||
defer fs.mu.Unlock()
|
||||
fs.garbage = body
|
||||
}
|
||||
|
||||
// SetDelay stalls every subsequent request for d before answering, so callers
|
||||
// can drive client timeouts and context cancellation deterministically. The
|
||||
// delay is abandoned as soon as the client hangs up.
|
||||
func (fs *fakeServer) SetDelay(d time.Duration) {
|
||||
fs.mu.Lock()
|
||||
defer fs.mu.Unlock()
|
||||
fs.delay = d
|
||||
}
|
||||
|
||||
// Count returns how many captured requests used the given method and path
|
||||
// prefix. "" matches any method.
|
||||
func (fs *fakeServer) Count(method, prefix string) int {
|
||||
n := 0
|
||||
for _, r := range fs.Requests() {
|
||||
if (method == "" || r.Method == method) && hasPrefix(r.Path, prefix) {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// Requests returns a snapshot of captured requests, in arrival order.
|
||||
func (fs *fakeServer) Requests() []capturedRequest {
|
||||
fs.mu.Lock()
|
||||
@@ -118,6 +172,32 @@ func fixtureNexusResolved(entityID, displayName, entityType string) string {
|
||||
})
|
||||
}
|
||||
|
||||
// fixtureNexusResolvedFlat is the flat resolve shape documented in
|
||||
// ECOSYSTEM-SPEC.md §1.5 (entity_id/entity_type/display_name at the top
|
||||
// level) rather than the nested "entity" object — the older of the two
|
||||
// wire shapes Maven must keep accepting.
|
||||
func fixtureNexusResolvedFlat(entityID, displayName, entityType string) string {
|
||||
return mustJSON(map[string]any{
|
||||
"status": "resolved",
|
||||
"entity_id": entityID,
|
||||
"entity_type": entityType,
|
||||
"display_name": displayName,
|
||||
})
|
||||
}
|
||||
|
||||
// fixtureNexusResolvedFuture is a resolved response from a hypothetical newer
|
||||
// Nexus: same required fields plus unknown ones. Decoding must ignore the
|
||||
// extras, not fail — forward compatibility is what lets the ecosystem be
|
||||
// upgraded one service at a time.
|
||||
func fixtureNexusResolvedFuture(entityID, displayName, entityType string) string {
|
||||
return mustJSON(map[string]any{
|
||||
"status": "resolved",
|
||||
"entity": map[string]any{"id": entityID, "display_name": displayName, "type": entityType, "tenant": "home"},
|
||||
"provenance": map[string]any{"resolver": "v3", "graph_epoch": 42},
|
||||
"score_breakdown": []any{map[string]any{"signal": "alias", "weight": 0.9}},
|
||||
})
|
||||
}
|
||||
|
||||
func fixtureNexusNotFound() string {
|
||||
return `{"status":"not_found"}`
|
||||
}
|
||||
@@ -138,6 +218,13 @@ func fixtureHexisExecuted(id, status string) string {
|
||||
return mustJSON(map[string]any{"id": id, "status": status})
|
||||
}
|
||||
|
||||
// fixtureHexisExecutionFailed is a well-formed Hexis response reporting that
|
||||
// the command itself failed: the call succeeded, the execution did not. Maven
|
||||
// must distinguish this from a transport failure and from success.
|
||||
func fixtureHexisExecutionFailed(id, message string) string {
|
||||
return mustJSON(map[string]any{"id": id, "status": "failed", "error": message})
|
||||
}
|
||||
|
||||
func fixturePraxisAttentionItems(items ...map[string]any) string {
|
||||
return mustJSON(items)
|
||||
}
|
||||
@@ -191,8 +278,13 @@ func newFakeNexus(t *testing.T, resolveBody string) *fakeServer {
|
||||
// fault is injected via SetFault.
|
||||
func newFakePraxis(t *testing.T, attentionBody string) *fakeServer {
|
||||
return newFakeServer(t, map[string]http.HandlerFunc{
|
||||
"GET /api/v1/tools/attention": jsonHandler(http.StatusOK, attentionBody),
|
||||
"POST /api/v1/tools/surface": jsonHandler(http.StatusOK, `{}`),
|
||||
"GET /api/v1/tools/attention": jsonHandler(http.StatusOK, attentionBody),
|
||||
"GET /api/v1/tools/changes": jsonHandler(http.StatusOK, `[]`),
|
||||
"POST /api/v1/tools/surface": jsonHandler(http.StatusOK, `{}`),
|
||||
"POST /api/v1/tools/acknowledge": jsonHandler(http.StatusOK, `{}`),
|
||||
"POST /api/v1/tools/resolve": jsonHandler(http.StatusOK, `{}`),
|
||||
"POST /api/v1/tools/ignore": jsonHandler(http.StatusOK, `{}`),
|
||||
"POST /api/v1/tools/pin": jsonHandler(http.StatusOK, `{}`),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/config"
|
||||
"github.com/kami/maven/internal/netscan"
|
||||
)
|
||||
|
||||
// scanBudget — the whole spoken scan, end to end. A voice turn that takes
|
||||
// longer than this has already failed as a turn, so the scan returns whatever
|
||||
// it found rather than keeping him waiting.
|
||||
const scanBudget = 20 * time.Second
|
||||
|
||||
// scanReadOut — how many hosts she names out loud. The rest are a count: a
|
||||
// spoken list of twenty IP addresses is not an answer.
|
||||
const scanReadOut = 6
|
||||
|
||||
// netWiring — the LAN scanner, when the `netscan` block is enabled. nil ⇒ Maven
|
||||
// never puts a discovery packet on the network.
|
||||
//
|
||||
// Unlike the house, a scan is a READ, so it is a query source rather than an
|
||||
// act: there is no allowlist row and no confirm turn, because nothing changes.
|
||||
// What makes that safe is that the range is not an argument — see
|
||||
// internal/netscan's package comment.
|
||||
type netWiring struct {
|
||||
scanner *netscan.Scanner
|
||||
subnets []string
|
||||
}
|
||||
|
||||
// wireNetScan builds the scanner. nil unless the block is enabled and valid.
|
||||
func wireNetScan(cfg *config.Config) *netWiring {
|
||||
nc, ok := cfg.NetScanner()
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if err := netscan.Validate(nc); err != nil {
|
||||
// config.validate already ran this, so reaching here is a programming
|
||||
// error rather than a config one. Not fatal: the scanner off is a
|
||||
// working Maven.
|
||||
log.Printf("netscan: not wired: %v", err)
|
||||
return nil
|
||||
}
|
||||
return &netWiring{scanner: netscan.New(nc), subnets: nc.Subnets}
|
||||
}
|
||||
|
||||
// scanSummary answers "какие устройства в сети?" in one line.
|
||||
func (w *netWiring) scanSummary(ctx context.Context) (string, bool) {
|
||||
if w == nil {
|
||||
return "", false
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, scanBudget)
|
||||
defer cancel()
|
||||
hosts, err := w.scanner.Scan(ctx)
|
||||
if err != nil {
|
||||
log.Printf("netscan: scan: %v", err)
|
||||
return "не получилось просканировать сеть.", true
|
||||
}
|
||||
if len(hosts) == 0 {
|
||||
return "в сети никого не нашла.", true
|
||||
}
|
||||
shown := hosts
|
||||
if len(shown) > scanReadOut {
|
||||
shown = shown[:scanReadOut]
|
||||
}
|
||||
parts := make([]string, 0, len(shown))
|
||||
for _, h := range shown {
|
||||
s := h.Addr
|
||||
if len(h.Ports) > 0 {
|
||||
ps := make([]string, 0, len(h.Ports))
|
||||
for _, p := range h.Ports {
|
||||
ps = append(ps, fmt.Sprintf("%d", p))
|
||||
}
|
||||
s += " (" + strings.Join(ps, ", ") + ")"
|
||||
}
|
||||
parts = append(parts, s)
|
||||
}
|
||||
out := fmt.Sprintf("нашла %d %s: %s", len(hosts), hostWord(len(hosts)), strings.Join(parts, "; "))
|
||||
if len(hosts) > len(shown) {
|
||||
out += fmt.Sprintf(" и ещё %d", len(hosts)-len(shown))
|
||||
}
|
||||
return out + ".", true
|
||||
}
|
||||
|
||||
// hostWord — Russian counts inflect the noun: 1 устройство, 2-4 устройства,
|
||||
// 5+ устройств, and the teens are all the last form.
|
||||
func hostWord(n int) string {
|
||||
if n%100 >= 11 && n%100 <= 14 {
|
||||
return "устройств"
|
||||
}
|
||||
switch n % 10 {
|
||||
case 1:
|
||||
return "устройство"
|
||||
case 2, 3, 4:
|
||||
return "устройства"
|
||||
default:
|
||||
return "устройств"
|
||||
}
|
||||
}
|
||||
|
||||
// isNetworkQuery recognises a question about the LAN, narrowly. It needs a
|
||||
// network word AND an ask: "интернет не работает" is a complaint, not a request
|
||||
// to scan, and a scan she runs unasked is exactly the noisy behaviour the
|
||||
// bounds exist to prevent.
|
||||
func isNetworkQuery(u string) bool {
|
||||
s := strings.ToLower(strings.TrimSpace(u))
|
||||
if s == "" {
|
||||
return false
|
||||
}
|
||||
network := false
|
||||
for _, w := range []string{"в сети", "в сетке", "сеть", "сети", "локальн", "wifi", "wi-fi", "вайфай"} {
|
||||
if strings.Contains(s, w) {
|
||||
network = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !network {
|
||||
return false
|
||||
}
|
||||
// An explicit ask to scan, or a phrase that can only be about the LAN.
|
||||
// "кто в сети" carries no device noun but means nothing else.
|
||||
for _, w := range []string{"просканируй", "сканируй", "скан", "просканир", "кто в сети", "кто в сетке"} {
|
||||
if strings.Contains(s, w) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
ask := strings.Contains(s, "?") || homeWord(s, "какие") || homeWord(s, "кто") ||
|
||||
homeWord(s, "что") || homeWord(s, "сколько") || strings.Contains(s, "покажи")
|
||||
if !ask {
|
||||
return false
|
||||
}
|
||||
for _, w := range []string{"устройств", "хост", "компьютер", "машин", "адрес"} {
|
||||
if strings.Contains(s, w) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/kami/maven/internal/config"
|
||||
)
|
||||
|
||||
func TestWireNetScanOffUnlessEnabled(t *testing.T) {
|
||||
for name, cfg := range map[string]*config.Config{
|
||||
"no block": {},
|
||||
"written but dark": {NetScan: &config.NetScanConfig{
|
||||
Subnets: []string{"192.168.1.0/24"},
|
||||
}},
|
||||
"enabled but nothing to scan": {NetScan: &config.NetScanConfig{Enabled: true}},
|
||||
"enabled but public": {NetScan: &config.NetScanConfig{
|
||||
Subnets: []string{"8.8.8.0/24"}, Enabled: true,
|
||||
}},
|
||||
"enabled but far too wide": {NetScan: &config.NetScanConfig{
|
||||
Subnets: []string{"10.0.0.0/8"}, Enabled: true,
|
||||
}},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
if w := wireNetScan(cfg); w != nil {
|
||||
t.Fatal("the scanner must not wire for this config")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
var w *netWiring
|
||||
if _, ok := w.scanSummary(context.Background()); ok {
|
||||
t.Fatal("a nil wiring must not claim a query")
|
||||
}
|
||||
|
||||
ok := wireNetScan(&config.Config{NetScan: &config.NetScanConfig{
|
||||
Subnets: []string{"192.168.1.0/24"}, Enabled: true,
|
||||
}})
|
||||
if ok == nil {
|
||||
t.Fatal("a valid enabled block should wire")
|
||||
}
|
||||
}
|
||||
|
||||
// A loopback /32 with nothing listening on the scanned port: the summary must
|
||||
// come back honest rather than inventing a host. This also exercises the real
|
||||
// dialer end to end without touching anything outside this box.
|
||||
func TestScanSummaryOnAnEmptyRange(t *testing.T) {
|
||||
w := wireNetScan(&config.Config{NetScan: &config.NetScanConfig{
|
||||
// Port 1 on loopback: nothing listens and the connection is refused
|
||||
// immediately, so the scan is fast and touches only this machine.
|
||||
Subnets: []string{"127.0.0.1/32"}, Ports: []int{1}, Rate: 1000, Enabled: true,
|
||||
}})
|
||||
if w == nil {
|
||||
t.Fatal("wireNetScan returned nil")
|
||||
}
|
||||
out, claimed := w.scanSummary(context.Background())
|
||||
if !claimed {
|
||||
t.Fatal("the summary did not claim the turn")
|
||||
}
|
||||
if out == "" {
|
||||
t.Fatal("empty summary")
|
||||
}
|
||||
// Persona: feminine self-reference, informal address, no pet names.
|
||||
low := strings.ToLower(out)
|
||||
for _, bad := range []string{"нашёл", "не смог ", "вы ", "ваш", "милый", "дорогой"} {
|
||||
if strings.Contains(low, bad) {
|
||||
t.Errorf("persona violation %q in %q", bad, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostWordAgreesWithTheCount(t *testing.T) {
|
||||
for n, want := range map[int]string{
|
||||
1: "устройство", 2: "устройства", 4: "устройства", 5: "устройств",
|
||||
11: "устройств", 12: "устройств", 21: "устройство", 22: "устройства",
|
||||
25: "устройств", 111: "устройств", 101: "устройство", 0: "устройств",
|
||||
} {
|
||||
if got := hostWord(n); got != want {
|
||||
t.Errorf("hostWord(%d) = %q, want %q", n, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsNetworkQuery(t *testing.T) {
|
||||
yes := []string{
|
||||
"какие устройства в сети?",
|
||||
"кто в сети?",
|
||||
"просканируй сеть",
|
||||
"покажи устройства в локальной сети",
|
||||
"сколько машин в сети",
|
||||
}
|
||||
no := []string{
|
||||
"",
|
||||
"интернет не работает",
|
||||
"сеть какая-то медленная",
|
||||
"я в сети инстаграма",
|
||||
"что включено дома?",
|
||||
"напомни оплатить интернет",
|
||||
}
|
||||
for _, u := range yes {
|
||||
if !isNetworkQuery(u) {
|
||||
t.Errorf("isNetworkQuery(%q) = false, want true", u)
|
||||
}
|
||||
}
|
||||
for _, u := range no {
|
||||
if isNetworkQuery(u) {
|
||||
t.Errorf("isNetworkQuery(%q) = true, want false", u)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -98,6 +98,11 @@ type reactiveHandler struct {
|
||||
// act allowlist and tool.Executor, like every other mutating act.
|
||||
home *homeWiring
|
||||
|
||||
// netscan — the LAN scanner (Vikunja #257). nil ⇒ off, which is the
|
||||
// default. A scan is a read, so it has no allowlist row; what keeps it
|
||||
// safe is that its range comes from config and from nowhere else.
|
||||
netscan *netWiring
|
||||
|
||||
weatherProvider weather.Provider
|
||||
weatherLocation string // default location for weather queries
|
||||
|
||||
|
||||
@@ -53,6 +53,9 @@ type voiceWiring struct {
|
||||
// enabled (Vikunja #256). Its devices land in the same allowlist as every
|
||||
// other act, so nothing else here has to know about it.
|
||||
home *homeWiring
|
||||
// netscan — the LAN scanner, nil unless the `netscan` block is enabled
|
||||
// (Vikunja #257).
|
||||
netscan *netWiring
|
||||
}
|
||||
|
||||
// close releases the listener + worker conns. Safe to call on nil (when
|
||||
@@ -161,6 +164,9 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, mem
|
||||
if w.home != nil {
|
||||
exec = exec.WithHome(w.home.caller())
|
||||
}
|
||||
// The LAN scanner (Vikunja #257): a read, bounded to the configured
|
||||
// subnets and rate-limited. Off unless the `netscan` block is enabled.
|
||||
w.netscan = wireNetScan(cfg)
|
||||
matcher := tool.NewMatcher(coreAPI)
|
||||
|
||||
// ----- weather provider (Open-Meteo when configured, Stub otherwise) -----
|
||||
@@ -245,6 +251,7 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, mem
|
||||
now: time.Now,
|
||||
feedsOn: cfg.Feeds != nil,
|
||||
home: w.home,
|
||||
netscan: w.netscan,
|
||||
// nil unless `crawl.on_demand` is on: reading a page he names is a
|
||||
// capability, and capabilities are off unless configured.
|
||||
crawler: onDemandCrawler(cfg),
|
||||
|
||||
@@ -56,6 +56,15 @@
|
||||
"enabled": false
|
||||
},
|
||||
|
||||
"netscan": {
|
||||
"subnets": ["192.168.1.0/24"],
|
||||
"ports": [22, 80, 443, 8080],
|
||||
"timeout": "400ms",
|
||||
"rate": 50,
|
||||
"max_hosts": 256,
|
||||
"enabled": false
|
||||
},
|
||||
|
||||
"nexus": { "url": "http://nexus:9740" },
|
||||
"praxis": { "url": "http://praxis:8989" },
|
||||
"hexis": { "url": "http://hexis:9741" },
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
"github.com/kami/maven/internal/delivery/telegramsink"
|
||||
"github.com/kami/maven/internal/mcp"
|
||||
"github.com/kami/maven/internal/morning"
|
||||
"github.com/kami/maven/internal/netscan"
|
||||
"github.com/kami/maven/internal/smarthome"
|
||||
"github.com/kami/maven/internal/update"
|
||||
"github.com/robfig/cron/v3"
|
||||
@@ -243,6 +244,11 @@ type Config struct {
|
||||
// disabled ⇒ Maven neither reads the house nor touches it, and no house row
|
||||
// exists in the act allowlist. See SmartHomeConfig.
|
||||
SmartHome *SmartHomeConfig `json:"smarthome,omitempty"`
|
||||
|
||||
// NetScan — the LAN scanner (Vikunja #257). nil / absent / disabled ⇒
|
||||
// Maven never puts a packet on the network looking for hosts. See
|
||||
// NetScanConfig.
|
||||
NetScan *NetScanConfig `json:"netscan,omitempty"`
|
||||
}
|
||||
|
||||
// MCPConfig — the MCP client block. Servers are dark until one has
|
||||
@@ -324,6 +330,51 @@ func (c *Config) SmartHomeClient() (smarthome.Config, bool) {
|
||||
}, true
|
||||
}
|
||||
|
||||
// NetScanConfig — the LAN scanner block (Vikunja #257). Dark until
|
||||
// `"enabled": true`.
|
||||
//
|
||||
// The important field is Subnets, and it is the ONLY source of a scan target.
|
||||
// Nothing an utterance, a router or a scanned host says can widen or move the
|
||||
// range: internal/netscan.Scanner.Scan takes no target argument at all. Each
|
||||
// subnet must be private and no larger than netscan.MaxPrefixHosts addresses
|
||||
// (a /22), enforced at config load rather than at the first spoken scan.
|
||||
type NetScanConfig struct {
|
||||
// Subnets — CIDRs to scan, "192.168.1.0/24".
|
||||
Subnets []string `json:"subnets,omitempty"`
|
||||
|
||||
// Ports — TCP ports to try per host. Empty ⇒ 22, 80, 443, 8080.
|
||||
Ports []int `json:"ports,omitempty"`
|
||||
|
||||
// Timeout — per-connection budget. 0 ⇒ 400ms.
|
||||
Timeout Duration `json:"timeout,omitempty"`
|
||||
|
||||
// Rate — connections per second across the whole scan. 0 ⇒ 50. Low on
|
||||
// purpose: a scan should look like background traffic, not a portscan.
|
||||
Rate int `json:"rate,omitempty"`
|
||||
|
||||
// MaxHosts — cap on addresses probed per scan. 0 ⇒ 256.
|
||||
MaxHosts int `json:"max_hosts,omitempty"`
|
||||
|
||||
// Enabled — false (the default) keeps a written block dark.
|
||||
Enabled bool `json:"enabled,omitempty"`
|
||||
}
|
||||
|
||||
// NetScanner maps the config block onto the netscan package's own type.
|
||||
// ok=false when absent or disabled, so validation and daemon wiring cannot
|
||||
// drift on the mapping.
|
||||
func (c *Config) NetScanner() (netscan.Config, bool) {
|
||||
if c.NetScan == nil || !c.NetScan.Enabled {
|
||||
return netscan.Config{}, false
|
||||
}
|
||||
return netscan.Config{
|
||||
Subnets: c.NetScan.Subnets,
|
||||
Ports: c.NetScan.Ports,
|
||||
Timeout: time.Duration(c.NetScan.Timeout),
|
||||
Rate: c.NetScan.Rate,
|
||||
MaxHosts: c.NetScan.MaxHosts,
|
||||
}, true
|
||||
}
|
||||
|
||||
// MCPServerConfig — one MCP server.
|
||||
type MCPServerConfig struct {
|
||||
// Name — the local handle. It prefixes every tool this server contributes
|
||||
@@ -1188,6 +1239,11 @@ func (c *Config) applyDefaults() {
|
||||
c.SmartHome.Refresh = Duration(DefaultSmartHomeRefresh)
|
||||
}
|
||||
|
||||
// Same rule for the scanner.
|
||||
if c.NetScan != nil && !c.NetScan.Enabled {
|
||||
c.NetScan = nil
|
||||
}
|
||||
|
||||
// Same rule for the crawler: a block that neither answers on demand nor
|
||||
// watches anything has nothing to do, so it is normalised to "off".
|
||||
if c.Crawl != nil && !c.Crawl.OnDemand && len(c.Crawl.Watches) == 0 {
|
||||
@@ -1308,6 +1364,13 @@ func (c *Config) validate() error {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// A scanner pointed at the public internet, or at a /8, fails here rather
|
||||
// than after the packets have already left.
|
||||
if nc, ok := c.NetScanner(); ok {
|
||||
if err := netscan.Validate(nc); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if len(c.MorningRoutines) > 0 {
|
||||
if err := morning.Validate(morningRoutinesFromConfig(c.MorningRoutines)); err != nil {
|
||||
return err
|
||||
|
||||
@@ -0,0 +1,356 @@
|
||||
// Package netscan discovers hosts on the LAN Maven is configured to look at
|
||||
// (Vikunja #257, docs/plans/12-bluetooth-network-scan.md).
|
||||
//
|
||||
// A scan is a read, but an unbounded scanner on a home network is noisy and is
|
||||
// trivially pointed somewhere it should not go, so the whole package is built
|
||||
// around four rules:
|
||||
//
|
||||
// - The target range NEVER comes from an utterance, a router, an LLM or a
|
||||
// device. Scan takes no target argument at all: it reads only the CIDRs in
|
||||
// the config block. There is deliberately no exported way to scan an
|
||||
// arbitrary range, so no amount of prompt injection or a rogue reply from a
|
||||
// scanned host can retarget it.
|
||||
// - Every configured CIDR must be private (RFC1918 / CGNAT / link-local) and
|
||||
// no larger than MaxPrefixHosts addresses. Scanning the public internet
|
||||
// from his flat is not a thing Maven does, and /8 is not a home LAN.
|
||||
// - Rate-limited. Connections leave at a fixed rate, so a scan looks like
|
||||
// background traffic rather than a portscan to anything watching.
|
||||
// - Bounded in total. MaxHosts, a per-connection timeout and the caller's
|
||||
// context all cap the work; a scan that runs long returns what it has.
|
||||
//
|
||||
// It is a TCP-connect scan (net.DialTimeout) and an ARP-table read. No raw
|
||||
// sockets, no SYN scan, no privileges: mavend does not run as root and this
|
||||
// does not ask it to.
|
||||
package netscan
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/netip"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultPorts — what a scan looks at when the config names nothing. Chosen to
|
||||
// answer "what is this box" on a home network, not to find a way in.
|
||||
var DefaultPorts = []int{22, 80, 443, 8080}
|
||||
|
||||
const (
|
||||
// DefaultTimeout — per-connection budget. Short: on a LAN a live host
|
||||
// answers in single-digit milliseconds, and a filtered port never answers.
|
||||
DefaultTimeout = 400 * time.Millisecond
|
||||
// DefaultRate — connections per second across the whole scan.
|
||||
DefaultRate = 50
|
||||
// DefaultMaxHosts — cap on addresses probed in one scan.
|
||||
DefaultMaxHosts = 256
|
||||
// MaxPrefixHosts — the largest CIDR that may be configured, in addresses.
|
||||
// 1024 is a /22: generous for a flat, and far short of anything that would
|
||||
// take minutes or wake up a neighbour's IDS.
|
||||
MaxPrefixHosts = 1024
|
||||
// maxParallel — in-flight dials. The rate limiter is the real throttle;
|
||||
// this only stops a slow subnet from piling up file descriptors.
|
||||
maxParallel = 16
|
||||
// arpFile — the kernel's ARP cache. Reading it is free and needs no packet.
|
||||
arpFile = "/proc/net/arp"
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrNotConfigured — no netscan block, or it is disabled.
|
||||
ErrNotConfigured = errors.New("netscan: not configured")
|
||||
// ErrNoSubnets — enabled with nothing to scan.
|
||||
ErrNoSubnets = errors.New("netscan: no subnets configured")
|
||||
)
|
||||
|
||||
// Config — the bounds of every scan. There is nothing here that can be
|
||||
// overridden at call time.
|
||||
type Config struct {
|
||||
// Subnets — the ONLY ranges that are ever probed, as CIDRs. Each must be
|
||||
// private and no bigger than MaxPrefixHosts.
|
||||
Subnets []string
|
||||
// Ports — TCP ports to try on each host. Empty ⇒ DefaultPorts.
|
||||
Ports []int
|
||||
// Timeout — per-connection budget. 0 ⇒ DefaultTimeout.
|
||||
Timeout time.Duration
|
||||
// Rate — connections per second. 0 ⇒ DefaultRate.
|
||||
Rate int
|
||||
// MaxHosts — cap on addresses probed per scan. 0 ⇒ DefaultMaxHosts.
|
||||
MaxHosts int
|
||||
}
|
||||
|
||||
// Host is one machine the scan saw.
|
||||
type Host struct {
|
||||
// Addr — the IP.
|
||||
Addr string
|
||||
// MAC — from the ARP cache, empty when the kernel has no entry.
|
||||
MAC string
|
||||
// Ports — open TCP ports, ascending.
|
||||
Ports []int
|
||||
}
|
||||
|
||||
// Up reports whether anything at all answered for this host.
|
||||
func (h Host) Up() bool { return len(h.Ports) > 0 || h.MAC != "" }
|
||||
|
||||
// Validate rejects a block that cannot safely run, at config-load time rather
|
||||
// than at the first spoken scan. This is the guard the whole package rides on:
|
||||
// if it passes, every later scan is inside these bounds by construction.
|
||||
func Validate(c Config) error {
|
||||
if len(c.Subnets) == 0 {
|
||||
return ErrNoSubnets
|
||||
}
|
||||
for _, s := range c.Subnets {
|
||||
p, err := netip.ParsePrefix(strings.TrimSpace(s))
|
||||
if err != nil {
|
||||
return fmt.Errorf("netscan: subnet %q: %w", s, err)
|
||||
}
|
||||
if !p.Addr().Is4() {
|
||||
return fmt.Errorf("netscan: subnet %q: only IPv4 is scanned", s)
|
||||
}
|
||||
if !isPrivate(p.Addr()) {
|
||||
return fmt.Errorf("netscan: subnet %q is not a private range: Maven does not scan the public internet", s)
|
||||
}
|
||||
if n := prefixHosts(p); n > MaxPrefixHosts {
|
||||
return fmt.Errorf("netscan: subnet %q covers %d addresses, limit is %d: narrow the prefix", s, n, MaxPrefixHosts)
|
||||
}
|
||||
}
|
||||
for _, port := range c.Ports {
|
||||
if port < 1 || port > 65535 {
|
||||
return fmt.Errorf("netscan: port %d out of range", port)
|
||||
}
|
||||
}
|
||||
if c.Rate < 0 || c.MaxHosts < 0 || c.Timeout < 0 {
|
||||
return errors.New("netscan: rate, max_hosts and timeout must not be negative")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// isPrivate — RFC1918, CGNAT and link-local. Loopback counts: scanning this box
|
||||
// is harmless and is how the tests run.
|
||||
func isPrivate(a netip.Addr) bool {
|
||||
if a.IsLoopback() || a.IsPrivate() || a.IsLinkLocalUnicast() {
|
||||
return true
|
||||
}
|
||||
// 100.64.0.0/10, the carrier-grade NAT range Tailscale hands out.
|
||||
cgnat := netip.MustParsePrefix("100.64.0.0/10")
|
||||
return cgnat.Contains(a)
|
||||
}
|
||||
|
||||
// prefixHosts — addresses covered by a v4 prefix.
|
||||
func prefixHosts(p netip.Prefix) int {
|
||||
bits := 32 - p.Bits()
|
||||
if bits >= 31 {
|
||||
return MaxPrefixHosts + 1
|
||||
}
|
||||
return 1 << bits
|
||||
}
|
||||
|
||||
// Scanner probes the configured subnets. Build it with New; the config it holds
|
||||
// is the config it was validated with, and nothing mutates it afterwards.
|
||||
type Scanner struct {
|
||||
cfg Config
|
||||
// dial is the connect seam; tests swap it.
|
||||
dial func(ctx context.Context, addr string, timeout time.Duration) bool
|
||||
// arp is the ARP-cache seam; tests swap it.
|
||||
arp func() (map[string]string, error)
|
||||
}
|
||||
|
||||
// New builds a scanner. Validate first — this does not.
|
||||
func New(cfg Config) *Scanner {
|
||||
if len(cfg.Ports) == 0 {
|
||||
cfg.Ports = append([]int(nil), DefaultPorts...)
|
||||
}
|
||||
if cfg.Timeout <= 0 {
|
||||
cfg.Timeout = DefaultTimeout
|
||||
}
|
||||
if cfg.Rate <= 0 {
|
||||
cfg.Rate = DefaultRate
|
||||
}
|
||||
if cfg.MaxHosts <= 0 {
|
||||
cfg.MaxHosts = DefaultMaxHosts
|
||||
}
|
||||
return &Scanner{cfg: cfg, dial: dialTCP, arp: readARP}
|
||||
}
|
||||
|
||||
// targets expands the configured subnets into addresses, skipping the network
|
||||
// and broadcast address of each, capped at MaxHosts. Deterministic order, so
|
||||
// two scans of an unchanged network read the same.
|
||||
func (s *Scanner) targets() []netip.Addr {
|
||||
var out []netip.Addr
|
||||
for _, cidr := range s.cfg.Subnets {
|
||||
p, err := netip.ParsePrefix(strings.TrimSpace(cidr))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
p = p.Masked()
|
||||
first := p.Addr()
|
||||
for a := first; p.Contains(a); a = a.Next() {
|
||||
if len(out) >= s.cfg.MaxHosts {
|
||||
return out
|
||||
}
|
||||
// Skip the network address; the broadcast address is skipped by
|
||||
// looking one ahead.
|
||||
if a == first && p.Bits() < 31 {
|
||||
continue
|
||||
}
|
||||
if p.Bits() < 31 && !p.Contains(a.Next()) {
|
||||
continue
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Scan probes every configured address and returns the hosts that answered.
|
||||
//
|
||||
// It takes no target: the range is the configured one, always. Callers pass a
|
||||
// context and nothing else, which is the point — see the package comment.
|
||||
func (s *Scanner) Scan(ctx context.Context) ([]Host, error) {
|
||||
if len(s.cfg.Subnets) == 0 {
|
||||
return nil, ErrNoSubnets
|
||||
}
|
||||
arp, err := s.arp()
|
||||
if err != nil {
|
||||
// A missing /proc/net/arp costs MAC addresses, not the scan.
|
||||
arp = map[string]string{}
|
||||
}
|
||||
|
||||
// One token per connection, at Rate per second, shared by every worker.
|
||||
interval := time.Second / time.Duration(s.cfg.Rate)
|
||||
if interval <= 0 {
|
||||
interval = time.Millisecond
|
||||
}
|
||||
tick := time.NewTicker(interval)
|
||||
defer tick.Stop()
|
||||
|
||||
type result struct {
|
||||
addr string
|
||||
ports []int
|
||||
}
|
||||
targets := s.targets()
|
||||
results := make(chan result, len(targets))
|
||||
sem := make(chan struct{}, maxParallel)
|
||||
var wg sync.WaitGroup
|
||||
|
||||
scan:
|
||||
for _, a := range targets {
|
||||
addr := a.String()
|
||||
for _, port := range s.cfg.Ports {
|
||||
// Checked before the select as well as inside it: select picks
|
||||
// randomly among ready cases, so at a high rate the ticker would
|
||||
// sometimes win over an already-canceled context and let one more
|
||||
// probe out.
|
||||
if ctx.Err() != nil {
|
||||
break scan
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break scan
|
||||
case <-tick.C:
|
||||
}
|
||||
sem <- struct{}{}
|
||||
wg.Add(1)
|
||||
go func(addr string, port int) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
if s.dial(ctx, net.JoinHostPort(addr, itoa(port)), s.cfg.Timeout) {
|
||||
results <- result{addr: addr, ports: []int{port}}
|
||||
}
|
||||
}(addr, port)
|
||||
}
|
||||
}
|
||||
wg.Wait()
|
||||
close(results)
|
||||
|
||||
byAddr := map[string]*Host{}
|
||||
for r := range results {
|
||||
h := byAddr[r.addr]
|
||||
if h == nil {
|
||||
h = &Host{Addr: r.addr}
|
||||
byAddr[r.addr] = h
|
||||
}
|
||||
h.Ports = append(h.Ports, r.ports...)
|
||||
}
|
||||
// A host in the ARP cache is up even with every port closed — it answered
|
||||
// an ARP request, which is the cheapest liveness signal there is.
|
||||
for _, a := range targets {
|
||||
addr := a.String()
|
||||
mac, ok := arp[addr]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if byAddr[addr] == nil {
|
||||
byAddr[addr] = &Host{Addr: addr}
|
||||
}
|
||||
byAddr[addr].MAC = mac
|
||||
}
|
||||
|
||||
out := make([]Host, 0, len(byAddr))
|
||||
for _, h := range byAddr {
|
||||
sort.Ints(h.Ports)
|
||||
out = append(out, *h)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
ai, _ := netip.ParseAddr(out[i].Addr)
|
||||
aj, _ := netip.ParseAddr(out[j].Addr)
|
||||
return ai.Less(aj)
|
||||
})
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func itoa(n int) string { return fmt.Sprintf("%d", n) }
|
||||
|
||||
func dialTCP(ctx context.Context, addr string, timeout time.Duration) bool {
|
||||
d := net.Dialer{Timeout: timeout}
|
||||
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
c, err := d.DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
_ = c.Close()
|
||||
return true
|
||||
}
|
||||
|
||||
func readARP() (map[string]string, error) {
|
||||
f, err := os.Open(arpFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
return parseARP(f)
|
||||
}
|
||||
|
||||
// parseARP reads the kernel's ARP table. Incomplete entries (all-zero MAC,
|
||||
// flags 0x0) are dropped: they mean "we asked and nobody answered", which is
|
||||
// the opposite of a discovered host.
|
||||
func parseARP(r io.Reader) (map[string]string, error) {
|
||||
out := map[string]string{}
|
||||
sc := bufio.NewScanner(r)
|
||||
first := true
|
||||
for sc.Scan() {
|
||||
if first { // header row
|
||||
first = false
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(sc.Text())
|
||||
if len(f) < 4 {
|
||||
continue
|
||||
}
|
||||
ip, flags, mac := f[0], f[2], f[3]
|
||||
if flags == "0x0" || mac == "00:00:00:00:00:00" {
|
||||
continue
|
||||
}
|
||||
if _, err := netip.ParseAddr(ip); err != nil {
|
||||
continue
|
||||
}
|
||||
out[ip] = mac
|
||||
}
|
||||
return out, sc.Err()
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
package netscan
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/netip"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestValidateBounds(t *testing.T) {
|
||||
ok := []Config{
|
||||
{Subnets: []string{"192.168.1.0/24"}},
|
||||
{Subnets: []string{"10.0.0.0/24", "172.16.5.0/28"}, Ports: []int{22, 80}},
|
||||
{Subnets: []string{"127.0.0.1/32"}},
|
||||
{Subnets: []string{"100.64.1.0/24"}}, // CGNAT / tailnet
|
||||
}
|
||||
for _, c := range ok {
|
||||
if err := Validate(c); err != nil {
|
||||
t.Errorf("Validate(%v) = %v, want nil", c.Subnets, err)
|
||||
}
|
||||
}
|
||||
|
||||
bad := map[string]Config{
|
||||
"nothing to scan": {},
|
||||
"public range": {Subnets: []string{"8.8.8.0/24"}},
|
||||
"whole internet": {Subnets: []string{"0.0.0.0/0"}},
|
||||
"a slash-8 is not a flat": {Subnets: []string{"10.0.0.0/8"}},
|
||||
"a /16 is too big": {Subnets: []string{"192.168.0.0/16"}},
|
||||
"not a cidr": {Subnets: []string{"192.168.1.1"}},
|
||||
"ipv6": {Subnets: []string{"fd00::/120"}},
|
||||
"garbage": {Subnets: []string{"выключи свет"}},
|
||||
"bad port": {Subnets: []string{"192.168.1.0/24"}, Ports: []int{0}},
|
||||
"huge port": {Subnets: []string{"192.168.1.0/24"}, Ports: []int{70000}},
|
||||
"negative rate": {Subnets: []string{"192.168.1.0/24"}, Rate: -1},
|
||||
}
|
||||
for name, c := range bad {
|
||||
if err := Validate(c); err == nil {
|
||||
t.Errorf("Validate(%s) = nil, want an error", strings.ReplaceAll(name, "\n", " "))
|
||||
}
|
||||
}
|
||||
if !errors.Is(Validate(Config{}), ErrNoSubnets) {
|
||||
t.Error("an empty block should report ErrNoSubnets")
|
||||
}
|
||||
}
|
||||
|
||||
// The whole safety story: a scanner probes its configured range and nothing
|
||||
// else. There is no API that takes a target, so this test asserts the negative
|
||||
// by watching every address the dialer was handed.
|
||||
func TestScanOnlyTouchesConfiguredSubnet(t *testing.T) {
|
||||
s := New(Config{Subnets: []string{"192.168.9.0/29"}, Ports: []int{80}, Rate: 10000})
|
||||
inside := netip.MustParsePrefix("192.168.9.0/29")
|
||||
|
||||
var mu sync.Mutex
|
||||
var seen []string
|
||||
s.dial = func(_ context.Context, addr string, _ time.Duration) bool {
|
||||
mu.Lock()
|
||||
seen = append(seen, addr)
|
||||
mu.Unlock()
|
||||
return addr == "192.168.9.3:80"
|
||||
}
|
||||
s.arp = func() (map[string]string, error) { return map[string]string{}, nil }
|
||||
|
||||
hosts, err := s.Scan(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Scan: %v", err)
|
||||
}
|
||||
if len(hosts) != 1 || hosts[0].Addr != "192.168.9.3" || len(hosts[0].Ports) != 1 {
|
||||
t.Fatalf("hosts = %+v", hosts)
|
||||
}
|
||||
// A /29 is 8 addresses; network (.0) and broadcast (.7) are skipped.
|
||||
if len(seen) != 6 {
|
||||
t.Errorf("probed %d addresses, want 6 (a /29 minus network and broadcast): %v", len(seen), seen)
|
||||
}
|
||||
for _, a := range seen {
|
||||
host, _, _ := strings.Cut(a, ":")
|
||||
ip, err := netip.ParseAddr(host)
|
||||
if err != nil || !inside.Contains(ip) {
|
||||
t.Errorf("probed %q, which is outside the configured subnet", a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestScanHonoursMaxHosts(t *testing.T) {
|
||||
s := New(Config{Subnets: []string{"192.168.9.0/24"}, Ports: []int{80}, Rate: 10000, MaxHosts: 3})
|
||||
var mu sync.Mutex
|
||||
n := 0
|
||||
s.dial = func(_ context.Context, _ string, _ time.Duration) bool {
|
||||
mu.Lock()
|
||||
n++
|
||||
mu.Unlock()
|
||||
return false
|
||||
}
|
||||
s.arp = func() (map[string]string, error) { return nil, nil }
|
||||
if _, err := s.Scan(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 3 {
|
||||
t.Errorf("dialed %d times, want 3 (MaxHosts)", n)
|
||||
}
|
||||
}
|
||||
|
||||
// The rate limiter must actually gate: 6 probes at 200/s cannot finish in less
|
||||
// than ~25ms. Asserted loosely, since a CI box is not a stopwatch.
|
||||
func TestScanIsRateLimited(t *testing.T) {
|
||||
s := New(Config{Subnets: []string{"192.168.9.0/29"}, Ports: []int{80}, Rate: 200})
|
||||
s.dial = func(context.Context, string, time.Duration) bool { return false }
|
||||
s.arp = func() (map[string]string, error) { return nil, nil }
|
||||
start := time.Now()
|
||||
if _, err := s.Scan(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if el := time.Since(start); el < 20*time.Millisecond {
|
||||
t.Errorf("6 probes at 200/s took %v: the rate limiter is not gating", el)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScanStopsOnCanceledContext(t *testing.T) {
|
||||
s := New(Config{Subnets: []string{"192.168.9.0/24"}, Ports: []int{80}, Rate: 10000})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
s.dial = func(context.Context, string, time.Duration) bool {
|
||||
t.Error("a canceled scan still dialed")
|
||||
return false
|
||||
}
|
||||
s.arp = func() (map[string]string, error) { return nil, nil }
|
||||
if _, err := s.Scan(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// A host with every port closed but an ARP entry is still up. A host outside
|
||||
// the configured range must not be reported even if the kernel knows it —
|
||||
// otherwise the ARP cache, which is populated by the network rather than by
|
||||
// Maven, would widen the answer past what he configured.
|
||||
func TestARPFillsMACWithinTheConfiguredRangeOnly(t *testing.T) {
|
||||
s := New(Config{Subnets: []string{"192.168.9.0/29"}, Ports: []int{80}, Rate: 10000})
|
||||
s.dial = func(context.Context, string, time.Duration) bool { return false }
|
||||
s.arp = func() (map[string]string, error) {
|
||||
return map[string]string{
|
||||
"192.168.9.2": "aa:bb:cc:dd:ee:ff",
|
||||
"10.9.9.9": "11:22:33:44:55:66",
|
||||
}, nil
|
||||
}
|
||||
hosts, err := s.Scan(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(hosts) != 1 {
|
||||
t.Fatalf("hosts = %+v", hosts)
|
||||
}
|
||||
if hosts[0].Addr != "192.168.9.2" || hosts[0].MAC != "aa:bb:cc:dd:ee:ff" {
|
||||
t.Errorf("host = %+v", hosts[0])
|
||||
}
|
||||
if !hosts[0].Up() {
|
||||
t.Error("an ARP entry with no open port is still a live host")
|
||||
}
|
||||
}
|
||||
|
||||
const arpFixture = `IP address HW type Flags HW address Mask Device
|
||||
192.168.1.1 0x1 0x2 3c:84:6a:11:22:33 * wlp1s0
|
||||
192.168.1.50 0x1 0x2 b8:27:eb:44:55:66 * wlp1s0
|
||||
192.168.1.77 0x1 0x0 00:00:00:00:00:00 * wlp1s0
|
||||
not-an-ip 0x1 0x2 de:ad:be:ef:00:01 * wlp1s0
|
||||
short line
|
||||
`
|
||||
|
||||
func TestParseARP(t *testing.T) {
|
||||
got, err := parseARP(strings.NewReader(arpFixture))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("got %d entries, want 2: %v", len(got), got)
|
||||
}
|
||||
if got["192.168.1.1"] != "3c:84:6a:11:22:33" || got["192.168.1.50"] != "b8:27:eb:44:55:66" {
|
||||
t.Errorf("entries = %v", got)
|
||||
}
|
||||
if _, ok := got["192.168.1.77"]; ok {
|
||||
t.Error("an incomplete ARP entry (flags 0x0) is not a discovered host")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewAppliesDefaults(t *testing.T) {
|
||||
s := New(Config{Subnets: []string{"192.168.1.0/24"}})
|
||||
if len(s.cfg.Ports) != len(DefaultPorts) || s.cfg.Rate != DefaultRate ||
|
||||
s.cfg.MaxHosts != DefaultMaxHosts || s.cfg.Timeout != DefaultTimeout {
|
||||
t.Errorf("defaults not applied: %+v", s.cfg)
|
||||
}
|
||||
// The defaults must not alias the package slice, or a second scanner could
|
||||
// rewrite DefaultPorts through it.
|
||||
s.cfg.Ports[0] = 9999
|
||||
if DefaultPorts[0] == 9999 {
|
||||
t.Error("New aliased DefaultPorts")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user