fix(ingest): name the event types and sources no detector reads, not unknown

README.md

@@ -6,7 +6,7 @@
 
 [![CI](https://github.com/TAIPANBOX/idryx/actions/workflows/ci.yml/badge.svg)](https://github.com/TAIPANBOX/idryx/actions/workflows/ci.yml)
 ![Go](https://img.shields.io/badge/go-1.27-00ADD8.svg)
-![tests](https://img.shields.io/badge/tests-374-brightgreen.svg)
+![tests](https://img.shields.io/badge/tests-381-brightgreen.svg)
 ![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)
 ![Status](https://img.shields.io/badge/phase-3%20%2B%20eBPF%20sensor-success.svg)
 
@@ -227,7 +227,9 @@ speak the [agent-passport](https://github.com/TAIPANBOX/agent-passport) spec:
   `sustained_loop`, `spend_spike`, `fanout_explosion`, `breaker_tripped`,
   `dlp_block`, `taint_block`, `mcp_drift`) and scopyx's two (`web_fetch`,
   `web_blocked`, which `unrouted_egress` reads) map to named constants; any
-  other type, from any producer, is tolerated generically.
+  other type, from any producer, is tolerated generically. Each ingest says on
+  stderr which event types and which sources no detector reads: they are
+  ingested as they are, and the line says so rather than calling them unknown.
   Every such stream is checked against the SPEC 6.5 `prev_hash` integrity
   chain on ingest, through `agent-stack-go`'s own `event.VerifyChain`, and
   the verdict goes to stderr: chain intact, chain broken at these lines, or

cmd/idryx/main.go

@@ -471,9 +471,8 @@ func populate(g *graph.Store, spec loadSpec) error {
 // message only; it plays no part in identity/event attribution, which
 // comes from each envelope's own `source` field).
 func reportTokenFuse(source, pathOrGlob string, rep tokenfuse.Report) {
-	if rep.Malformed > 0 || len(rep.UnknownTypes) > 0 {
-		fmt.Fprintf(os.Stderr, "idryx: %s %s: %d line(s) read, %d malformed, %d unknown event type(s)\n",
-			source, pathOrGlob, rep.Lines, rep.Malformed, len(rep.UnknownTypes))
+	if msg := tokenfuse.StderrSummary(rep); msg != "" {
+		fmt.Fprintf(os.Stderr, "idryx: %s %s: %s\n", source, pathOrGlob, msg)
 	}
 	reportStreams(source, pathOrGlob, rep)
 	reportChain(source, pathOrGlob, rep.Chain)

cmd/idryx/main_test.go

@@ -1097,3 +1097,43 @@ func TestAIInventoryReadsAQryxDocument(t *testing.T) {
 		t.Errorf("root = %q, want /src", scan.Root)
 	}
 }
+
+// TestARegisteredTypeNoDetectorReadsIsNotCalledUnknown is the operator-facing
+// half of the unread-type fix. Four types the bus registry names (agent-passport
+// SPEC 6.2) are ingested as they are, and no idryx detector reads them yet. The
+// stderr line must say exactly that, and must not call a registered type
+// "unknown", which is what it said before.
+func TestARegisteredTypeNoDetectorReadsIsNotCalledUnknown(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "tokenfuse.ndjson")
+	var lines []string
+	for i, typ := range []string{"spend_spike", "identity_mismatch", "taint_raised", "unit_cap_exceeded", "dependency_failed"} {
+		lines = append(lines, fmt.Sprintf(`{"schema":%q,"ts":"2026-08-05T10:0%d:00Z","source":"tokenfuse","type":%q,"agent_id":"agent://acme.example/bot"}`,
+			event.SchemaV02, i, typ))
+	}
+	if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	out := captureStderr(t, func() {
+		if _, err := buildGraph("", "", "", "", "", "", "", loadList{{Source: "tokenfuse", Path: path}}); err != nil {
+			t.Fatalf("buildGraph: %v", err)
+		}
+	})
+	if strings.Contains(out, "unknown") {
+		t.Errorf("a registered type must not be called unknown, got %q", out)
+	}
+	if !strings.Contains(out, "4 event type(s) no idryx detector reads, ingested as they are") {
+		t.Errorf("the four unread types must be named as ingested as they are, got %q", out)
+	}
+	for _, typ := range []string{"identity_mismatch", "taint_raised", "unit_cap_exceeded", "dependency_failed"} {
+		if !strings.Contains(out, fmt.Sprintf("%q", typ)) {
+			t.Errorf("the line must name %s, got %q", typ, out)
+		}
+	}
+	if strings.Contains(out, `"spend_spike"`) {
+		t.Errorf("a type a detector reads must not be listed, got %q", out)
+	}
+	if strings.Contains(out, "source(s) no idryx detector reads") {
+		t.Errorf("tokenfuse has a type a detector reads, so it must not be listed as an unread source, got %q", out)
+	}
+}

features/unread-is-named-not-unknown.feature

@@ -0,0 +1,32 @@
+Feature: An ingest names the event types and sources no detector reads, and does not call a registered type unknown
+
+  The shared event bus registry names event types that no idryx detector reads
+  yet. The stderr line used to count every such type as unknown, which is false
+  for a type the registry names. Those events are still ingested as they are,
+  carried through and counted, never dropped. The line now says exactly that:
+  it names the types and the sources no detector reads, in a bounded list, and
+  it stays silent when every type and source is one a detector reads.
+
+  Scenario: a registered type no detector reads is named as ingested as it is
+    Given tokenfuse.ndjson holds a spend event and events of four registered types that no detector reads
+    When idryx loads the file
+    Then stderr names the four types as event types no detector reads, ingested as they are, and calls none of them unknown
+  # @test:TestARegisteredTypeNoDetectorReadsIsNotCalledUnknown
+
+  Scenario: a source none of whose events a detector reads is named, across every file of a load
+    Given typryx.ndjson holds events of a type no detector reads, and tokenfuse is read in one file of the load and not in the other
+    When idryx loads the glob over both
+    Then stderr names typryx as a source no detector reads, and does not name tokenfuse
+  # @test:TestAnUnreadSourceIsJudgedAcrossEveryFileOfALoad
+
+  Scenario: a long or hostile name stays on one bounded line
+    Given a load carries more than eight unread type names, one of them over sixty-four bytes and one holding a line break
+    When the stderr summary of the load is written
+    Then it is one line, the hostile name is escaped, the long name is cut, eight names are listed and the rest are counted
+  # @test:TestStderrSummaryBoundsAndEscapesEveryName
+
+  Scenario: an ingest that a detector reads in full says nothing
+    Given a load holds only types and sources that a detector reads
+    When the stderr summary of the load is written
+    Then the summary is empty
+  # @test:TestStderrSummaryIsEmptyWhenEveryTypeAndSourceIsRead

internal/ingest/tokenfuse/tokenfuse.go

@@ -77,6 +77,11 @@ type Report struct {
 	Malformed    int
 	UnknownTypes map[string]int
 	Chain        Chain
+	// Sources tallies each source this ingest carried: how many of its events
+	// were ingested, and how many of those carry a type a detector reads (a
+	// type in knownTypes). A source none of whose events is read is one no
+	// detector reads, and UnreadSources names it.
+	Sources map[string]SourceTally
 
 	// ForeignSource counts well-formed events whose `source` the file they
 	// were read from may not carry. They are NOT ingested: no identity, no
@@ -97,6 +102,14 @@ type Report struct {
 	Unknown []UnknownFile
 }
 
+// SourceTally is what one source contributed to a Report.
+type SourceTally struct {
+	// Events is how many of the source's events were ingested.
+	Events int
+	// Read is how many of them carry an event type in knownTypes.
+	Read int
+}
+
 // Foreign is one (file, claimed source) pair whose events were refused.
 type Foreign struct {
 	File, Stem string
@@ -179,7 +192,78 @@ func (c Chain) Present() bool {
 func (c Chain) Intact() bool { return c.Verified && len(c.Breaks) == 0 }
 
 func newReport() Report {
-	return Report{UnknownTypes: map[string]int{}}
+	return Report{UnknownTypes: map[string]int{}, Sources: map[string]SourceTally{}}
+}
+
+// UnreadSources lists, sorted, the sources this ingest carried none of whose
+// events a detector reads: each has ingested events, and none of them is a
+// type in knownTypes. A source read in one file of a load is read for the
+// whole load.
+func (r Report) UnreadSources() []string {
+	var out []string
+	for s, t := range r.Sources {
+		if t.Events > 0 && t.Read == 0 {
+			out = append(out, s)
+		}
+	}
+	sort.Strings(out)
+	return out
+}
+
+// maxNamed bounds how many names one list in StderrSummary shows, and
+// maxNameShown how many bytes of one name, before it is quoted.
+const (
+	maxNamed     = 8
+	maxNameShown = 64
+)
+
+// StderrSummary is the body of the one stderr line an agent-event-bus ingest
+// earns from its own counts, after the caller's "idryx: <source> <path>:"
+// prefix, or "" when there is nothing to say. It names the malformed lines,
+// the event types no detector reads and the sources no detector reads. Those
+// are ingested as they are (carried through, counted, never dropped), so the
+// line says exactly that and never calls a registered type unknown.
+func StderrSummary(rep Report) string {
+	var parts []string
+	if rep.Malformed > 0 {
+		parts = append(parts, fmt.Sprintf("%d line(s) read, %d malformed", rep.Lines, rep.Malformed))
+	}
+	if len(rep.UnknownTypes) > 0 {
+		types := make([]string, 0, len(rep.UnknownTypes))
+		for t := range rep.UnknownTypes {
+			types = append(types, t)
+		}
+		sort.Strings(types)
+		parts = append(parts, fmt.Sprintf("%d event type(s) no idryx detector reads, ingested as they are: %s",
+			len(types), namesList(types)))
+	}
+	if srcs := rep.UnreadSources(); len(srcs) > 0 {
+		parts = append(parts, fmt.Sprintf("%d source(s) no idryx detector reads, ingested as they are: %s",
+			len(srcs), namesList(srcs)))
+	}
+	return strings.Join(parts, "; ")
+}
+
+// namesList quotes the first maxNamed of names, which are sorted already, and
+// says how many it did not show. A producer wrote these words: each is cut and
+// then escaped by %q, so nothing in one can break the line it is printed on.
+func namesList(names []string) string {
+	shown := names
+	if len(shown) > maxNamed {
+		shown = shown[:maxNamed]
+	}
+	quoted := make([]string, len(shown))
+	for i, n := range shown {
+		if len(n) > maxNameShown {
+			n = n[:maxNameShown] + "..."
+		}
+		quoted[i] = fmt.Sprintf("%q", n)
+	}
+	out := strings.Join(quoted, ", ")
+	if rest := len(names) - len(shown); rest > 0 {
+		out += fmt.Sprintf(" +%d more", rest)
+	}
+	return out
 }
 
 func (r *Report) merge(o Report) {
@@ -188,6 +272,12 @@ func (r *Report) merge(o Report) {
 	for t, n := range o.UnknownTypes {
 		r.UnknownTypes[t] += n
 	}
+	for s, t := range o.Sources {
+		cur := r.Sources[s]
+		cur.Events += t.Events
+		cur.Read += t.Read
+		r.Sources[s] = cur
+	}
 	r.Chain.merge(o.Chain)
 	r.ForeignSource += o.ForeignSource
 	r.UnknownStream += o.UnknownStream
@@ -419,6 +509,12 @@ func parse(data []byte, file string, pol *stream.Policy) ([]model.Identity, []mo
 			evType = model.EventType(env.Type) // generic: pass the raw type through, never error
 			rep.UnknownTypes[env.Type]++
 		}
+		tally := rep.Sources[env.Source]
+		tally.Events++
+		if known {
+			tally.Read++
+		}
+		rep.Sources[env.Source] = tally
 		events = append(events, model.Event{
 			Time:       ts,
 			IdentityID: env.AgentID,

internal/ingest/tokenfuse/tokenfuse_test.go

@@ -537,3 +537,134 @@ func TestTheMergedReportBoundsTheNamedPairsAcrossFiles(t *testing.T) {
 		t.Fatalf("UnknownStream = %d, named = %d, want %d and %d", urep.UnknownStream, len(urep.Unknown), maxForeignPairs+5, maxForeignPairs)
 	}
 }
+
+// busLine is one agent-event envelope as a producer writes it. It carries no
+// prev_hash, so the chain verdict is "no chain" and stays out of the way.
+func busLine(source, typ, agent string) string {
+	return fmt.Sprintf(`{"schema":"taipanbox.dev/agent-event/v0.2","ts":"2026-08-05T10:00:00Z","source":%q,"type":%q,"agent_id":%q}`,
+		source, typ, agent)
+}
+
+// TestStderrSummaryNamesTheTypesNoDetectorReads: the types nothing reads are
+// named as ingested as they are, sorted, never called unknown.
+func TestStderrSummaryNamesTheTypesNoDetectorReads(t *testing.T) {
+	rep := newReport()
+	rep.Lines = 3
+	rep.UnknownTypes["taint_raised"] = 1
+	rep.UnknownTypes["identity_mismatch"] = 2
+	want := `2 event type(s) no idryx detector reads, ingested as they are: "identity_mismatch", "taint_raised"`
+	if got := StderrSummary(rep); got != want {
+		t.Errorf("StderrSummary = %q, want %q", got, want)
+	}
+}
+
+// TestStderrSummaryIsEmptyWhenEveryTypeAndSourceIsRead: an ingest a detector
+// reads in full says nothing, as the malformed count says nothing when zero.
+func TestStderrSummaryIsEmptyWhenEveryTypeAndSourceIsRead(t *testing.T) {
+	rep := newReport()
+	rep.Lines = 3
+	rep.Sources["tokenfuse"] = SourceTally{Events: 3, Read: 3}
+	if got := StderrSummary(rep); got != "" {
+		t.Errorf("a clean ingest must say nothing, got %q", got)
+	}
+}
+
+// TestStderrSummaryJoinsItsPartsOnOneLine: malformed lines, unread types and
+// unread sources share one line, separated by semicolons, in that order.
+func TestStderrSummaryJoinsItsPartsOnOneLine(t *testing.T) {
+	rep := newReport()
+	rep.Lines, rep.Malformed = 5, 1
+	rep.UnknownTypes["dependency_failed"] = 1
+	rep.Sources["typryx"] = SourceTally{Events: 1, Read: 0}
+	want := `5 line(s) read, 1 malformed; 1 event type(s) no idryx detector reads, ingested as they are: "dependency_failed"; ` +
+		`1 source(s) no idryx detector reads, ingested as they are: "typryx"`
+	if got := StderrSummary(rep); got != want {
+		t.Errorf("StderrSummary = %q, want %q", got, want)
+	}
+}
+
+// TestStderrSummaryBoundsAndEscapesEveryName: a producer writes these words.
+// Whatever it wrote stays on one line, a long name is cut at 64 bytes, and
+// only eight names are listed, with the rest counted.
+func TestStderrSummaryBoundsAndEscapesEveryName(t *testing.T) {
+	rep := newReport()
+	long := "a" + strings.Repeat("x", 100)
+	rep.UnknownTypes[long] = 1
+	rep.UnknownTypes["bad\nline\x1b[31m"] = 1
+	for i := 0; i < 10; i++ {
+		rep.UnknownTypes[fmt.Sprintf("type_%02d", i)] = 1
+	}
+	got := StderrSummary(rep)
+	if strings.ContainsAny(got, "\n\x1b") {
+		t.Errorf("the line must hold no raw line break or control byte, got %q", got)
+	}
+	// 64 bytes in all, the leading "a" included, then the marker.
+	if !strings.Contains(got, `"a`+strings.Repeat("x", 63)+`..."`) {
+		t.Errorf("a long name must be cut at 64 bytes and marked, got %q", got)
+	}
+	if strings.Contains(got, strings.Repeat("x", 64)) {
+		t.Errorf("a long name must not be shown past 64 bytes, got %q", got)
+	}
+	if !strings.Contains(got, `"bad\nline\x1b[31m"`) {
+		t.Errorf("a hostile name must be escaped, got %q", got)
+	}
+	if !strings.Contains(got, `"type_05" +4 more`) {
+		t.Errorf("eight names then the count of the rest, got %q", got)
+	}
+	if strings.Contains(got, "type_06") {
+		t.Errorf("a ninth name must not be listed, got %q", got)
+	}
+	if !strings.Contains(got, "12 event type(s) no idryx detector reads") {
+		t.Errorf("the count must be the whole, got %q", got)
+	}
+}
+
+// TestStderrSummaryNamesMalformedLinesOnly: a malformed line alone says how
+// many lines were read and how many were malformed, and nothing about types.
+func TestStderrSummaryNamesMalformedLinesOnly(t *testing.T) {
+	rep := newReport()
+	rep.Lines, rep.Malformed = 5, 1
+	rep.Sources["tokenfuse"] = SourceTally{Events: 4, Read: 4}
+	if got, want := StderrSummary(rep), "5 line(s) read, 1 malformed"; got != want {
+		t.Errorf("StderrSummary = %q, want %q", got, want)
+	}
+}
+
+// TestAnUnreadSourceIsJudgedAcrossEveryFileOfALoad: a source is read when any
+// of its events, in any file of the load, carries a type a detector reads. A
+// line refused by the stream rule is not a source the file claimed to carry.
+func TestAnUnreadSourceIsJudgedAcrossEveryFileOfALoad(t *testing.T) {
+	dir := t.TempDir()
+	write := func(name string, lines ...string) {
+		t.Helper()
+		if err := os.WriteFile(filepath.Join(dir, name), []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// tokenfuse reads one type in one file and not in the other.
+	write("tokenfuse.ndjson", busLine("tokenfuse", "identity_mismatch", "agent://acme.example/a"))
+	write("tokenfuse-cloud.ndjson", busLine("tokenfuse", "spend_spike", "agent://acme.example/a"))
+	// typryx reads nothing, and its line claiming wardryx is refused.
+	write("typryx.ndjson",
+		busLine("typryx", "answer_given", "agent://acme.example/t"),
+		busLine("wardryx", "policy_deny", "agent://acme.example/t"))
+
+	_, _, rep, err := Load(filepath.Join(dir, "*.ndjson"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := rep.UnreadSources(); len(got) != 1 || got[0] != "typryx" {
+		t.Fatalf("UnreadSources = %v, want [typryx]", got)
+	}
+	if got, want := rep.Sources["tokenfuse"], (SourceTally{Events: 2, Read: 1}); got != want {
+		t.Errorf("tokenfuse tally = %+v, want %+v", got, want)
+	}
+	if _, counted := rep.Sources["wardryx"]; counted {
+		t.Errorf("a refused line must not be tallied under the source it claimed: %v", rep.Sources)
+	}
+	want := `2 event type(s) no idryx detector reads, ingested as they are: "answer_given", "identity_mismatch"; ` +
+		`1 source(s) no idryx detector reads, ingested as they are: "typryx"`
+	if got := StderrSummary(rep); got != want {
+		t.Errorf("StderrSummary = %q, want %q", got, want)
+	}
+}

scripts/gates-have-teeth.sh

@@ -479,10 +479,14 @@ run_case "features-are-bound: prose edited inside a scenario" pass \
 	'./scripts/features-are-bound.sh' \
 	"$(py 'edit("features/source-must-match-its-stream.feature", "Given tokenfuse.ndjson holds a line claiming wardryx", "Given tokenfuse.ndjson holds one line claiming wardryx (a harmless prose edit)")')"
 
+# Every feature file goes, not one named file: the case means "no scenario left
+# in features/", and a second feature file (added after this case was written)
+# would otherwise keep scenarios alive and turn the gate's silence into a pass.
 run_case "features-are-bound: no scenario left, so it measured nothing" fail \
 	'./scripts/features-are-bound.sh' \
-	"$(py 'import os
-os.remove("features/source-must-match-its-stream.feature")')" \
+	"$(py 'import glob, os
+for f in glob.glob("features/*.feature"):
+    os.remove(f)')" \
 	"measured nothing"
 
 echo