c4ec7226 — eitri

docs/openapi.json

Old		New
@@ -322,6 +322,16 @@
322	"scope": {	322	"scope": {
323	"type": "string"	323	"type": "string"
324	},	324	},
		325	"sessions": {
		326	"anyOf": [
		327	{
		328	"$ref": "#/components/schemas/ExposureSessions"
		329	},
		330	{
		331	"type": "null"
		332	}
		333	]
		334	},
325	"state": {	335	"state": {
326	"type": "string"	336	"type": "string"
327	},	337	},
@@ -344,6 +354,25 @@
344	],	354	],
345	"type": "object"	355	"type": "object"
346	},	356	},
		357	"ExposureSessions": {
		358	"properties": {
		359	"active": {
		360	"type": "integer"
		361	},
		362	"dropped": {
		363	"type": "integer"
		364	},
		365	"refused": {
		366	"type": "integer"
		367	}
		368	},
		369	"required": [
		370	"active",
		371	"dropped",
		372	"refused"
		373	],
		374	"type": "object"
		375	},
347	"Host": {	376	"Host": {
348	"properties": {	377	"properties": {
349	"agent_update_available": {	378	"agent_update_available": {
@@ -778,6 +807,9 @@
778	"status": {	807	"status": {
779	"type": "string"	808	"type": "string"
780	},	809	},
		810	"status_detail": {
		811	"type": "string"
		812	},
781	"trusted_cas": {	813	"trusted_cas": {
782	"items": {	814	"items": {
783	"$ref": "#/components/schemas/TrustedCA"	815	"$ref": "#/components/schemas/TrustedCA"
@@ -808,6 +840,7 @@
808	"phase",	840	"phase",
809	"power_state",	841	"power_state",
810	"status",	842	"status",
		843	"status_detail",
811	"vcpus"	844	"vcpus"
812	],	845	],
813	"type": "object"	846	"type": "object"

internal/agent/exposeproxy/exposeproxy.go

Old		New
@@ -111,6 +111,15 @@ type exposure struct {
111	reason string	111	reason string
112	conns atomic.Int64	112	conns atomic.Int64
113		113
		114	// refused and dropped are running totals for the life of this entry, which
		115	// is the life of the exposure on this agent: an id the fleet still wants
		116	// keeps its entry across rebinds, and an agent restart rebuilds every socket
		117	// from the first snapshot and starts both at zero. Totals rather than
		118	// gauges because what they count is momentary — see ExposureSessions in the
		119	// proto.
		120	refused atomic.Int64
		121	dropped atomic.Int64
		122
114	// sessions is the UDP session table, guarded by its own mutex because the	123	// sessions is the UDP session table, guarded by its own mutex because the
115	// packet loop touches it on every datagram while the manager lock is held	124	// packet loop touches it on every datagram while the manager lock is held
116	// across whole converges. nil until the first UDP bind.	125	// across whole converges. nil until the first UDP bind.
@@ -118,6 +127,23 @@ type exposure struct {
118	sessions map[string]*udpSession	127	sessions map[string]*udpSession
119	}	128	}
120		129
		130	// counters is what this exposure has carried, for the report. active is read
		131	// from whichever half of the pipe this exposure is: a TCP exposure holds
		132	// connections, a UDP one holds sessions, and no exposure is both.
		133	func (e exposure) counters() pb.ExposureSessions {
		134	active := e.conns.Load()
		135	if e.pc != nil {
		136	e.smu.Lock()
		137	active = int64(len(e.sessions))
		138	e.smu.Unlock()
		139	}
		140	return &pb.ExposureSessions{
		141	Active: active,
		142	Refused: e.refused.Load(),
		143	Dropped: e.dropped.Load(),
		144	}
		145	}
		146
121	// bound reports whether this exposure currently holds a socket.	147	// bound reports whether this exposure currently holds a socket.
122	func (e *exposure) bound() bool { return e.ln != nil \|\| e.pc != nil }	148	func (e *exposure) bound() bool { return e.ln != nil \|\| e.pc != nil }
123		149
@@ -188,7 +214,7 @@ func (m Manager) Converge(desired []pb.ExposureDesired) []*pb.ExposureActual {
188	if !ex.bound() {	214	if !ex.bound() {
189	if err := m.bind(d, ex); err != nil {	215	if err := m.bind(d, ex); err != nil {
190	ex.reason = err.Error()	216	ex.reason = err.Error()
191	out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "failed", Reason: ex.reason})	217	out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "failed", Reason: ex.reason, Sessions: ex.counters()})
192	continue	218	continue
193	}	219	}
194	}	220	}
@@ -201,7 +227,7 @@ func (m Manager) Converge(desired []pb.ExposureDesired) []*pb.ExposureActual {
201	if ex.pc != nil {	227	if ex.pc != nil {
202	ex.evictMovedSessions(m.addr(d.GetVmId()))	228	ex.evictMovedSessions(m.addr(d.GetVmId()))
203	}	229	}
204	out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "active", Reason: ex.reason})	230	out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "active", Reason: ex.reason, Sessions: ex.counters()})
205	}	231	}
206	return out	232	return out
207	}	233	}
@@ -303,7 +329,9 @@ func (m Manager) accept(id string, ex exposure, ln net.Listener, vmID string,
303	// a stale count and all pass.	329	// a stale count and all pass.
304	if ex.conns.Add(1) > m.maxConns {	330	if ex.conns.Add(1) > m.maxConns {
305	ex.conns.Add(-1)	331	ex.conns.Add(-1)
		332	ex.refused.Add(1)
306	conn.Close()	333	conn.Close()
		334	slog.Debug("exposure refused a connection at its cap", "exposure", id, "cap", m.maxConns, "client", conn.RemoteAddr())
307	if !capped {	335	if !capped {
308	capped = true	336	capped = true
309	slog.Warn("exposure at its connection cap, refusing", "exposure", id, "cap", m.maxConns)	337	slog.Warn("exposure at its connection cap, refusing", "exposure", id, "cap", m.maxConns)
@@ -311,9 +339,11 @@ func (m Manager) accept(id string, ex exposure, ln net.Listener, vmID string,
311	continue	339	continue
312	}	340	}
313	capped = false	341	capped = false
		342	slog.Debug("exposure connection opened", "exposure", id, "client", conn.RemoteAddr())
314	go func() {	343	go func() {
315	defer ex.conns.Add(-1)	344	defer ex.conns.Add(-1)
316	m.pipe(conn, vmID, guestPort)	345	m.pipe(ex, conn, vmID, guestPort)
		346	slog.Debug("exposure connection closed", "exposure", id, "client", conn.RemoteAddr())
317	}()	347	}()
318	}	348	}
319	}	349	}
@@ -343,15 +373,19 @@ func outOfDescriptors(err error) bool {
343	// pipe connects one accepted connection to the guest. No address (a guest still	373	// pipe connects one accepted connection to the guest. No address (a guest still
344	// leasing) or a guest that will not answer closes immediately: the host half of	374	// leasing) or a guest that will not answer closes immediately: the host half of
345	// the pipe exists, the guest half does not, and waiting would only hold the	375	// the pipe exists, the guest half does not, and waiting would only hold the
346	// caller open on a promise nothing is keeping.	376	// caller open on a promise nothing is keeping. Both are counted as drops — from
347	func (m *Manager) pipe(client net.Conn, vmID string, guestPort uint32) {	377	// the caller's side the port accepted and then said nothing, and the exposure's
		378	// own totals are the only place that difference is written down.
		379	func (m Manager) pipe(ex exposure, client net.Conn, vmID string, guestPort uint32) {
348	ip := m.addr(vmID)	380	ip := m.addr(vmID)
349	if ip == "" {	381	if ip == "" {
		382	ex.dropped.Add(1)
350	client.Close()	383	client.Close()
351	return	384	return
352	}	385	}
353	guest, err := net.Dial("tcp", net.JoinHostPort(ip, strconv.Itoa(int(guestPort))))	386	guest, err := net.Dial("tcp", net.JoinHostPort(ip, strconv.Itoa(int(guestPort))))
354	if err != nil {	387	if err != nil {
		388	ex.dropped.Add(1)
355	client.Close()	389	client.Close()
356	return	390	return
357	}	391	}

internal/agent/exposeproxy/exposeproxy_test.go

@@ @@ -593,3 +593,96 @@ func TestStopAllClosesEveryListener(t *testing.T) { @@
+		assert.Error(t, err, "listener %s survived StopAll", addr)
+// counters converges the same desired set again and returns what the named
+// exposure reports about what it has carried. Converge is level-triggered, so
+// asking twice is how the report is taken in production too.
+func counters(t *testing.T, m *Manager, d []*pb.ExposureDesired, id string) *pb.ExposureSessions {
+	t.Helper()
+	for _, a := range m.Converge(d) {
+		if a.GetId() == id {
+			require.NotNil(t, a.GetSessions(), "exposure %q reported no counters", id)
+			return a.GetSessions()
+		}
+	}
+	t.Fatalf("exposure %q not in the report", id)
+	return nil
+}
+// TestExposureCountsWhatItHoldsAndWhatItTurnsAway is the observability leg: a
+// published port that is quietly refusing callers looks exactly like a healthy
+// one from outside, and these three numbers are the difference.
+func TestExposureCountsWhatItHoldsAndWhatItTurnsAway(t *testing.T) {
+	g := newHoldingGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	m.maxConns = 1
+	spec := []*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)}
+	m.Converge(spec)
+	addr := boundPort(t, m, "e1")
+	held := hold(t, addr)
+	waitConns(t, m, "e1", 1)
+	got := counters(t, m, spec, "e1")
+	assert.Equal(t, int64(1), got.GetActive())
+	assert.Zero(t, got.GetRefused(), "nobody has been turned away yet")
+	// One caller past the cap, refused and closed.
+	over, err := net.Dial("tcp", addr)
+	require.NoError(t, err)
+	defer over.Close()
+	require.NoError(t, over.SetDeadline(time.Now().Add(5*time.Second)))
+	_, err = io.ReadAll(over)
+	require.NoError(t, err)
+	got = counters(t, m, spec, "e1")
+	assert.Equal(t, int64(1), got.GetActive(), "the refusal did not disturb the connection under the cap")
+	assert.Equal(t, int64(1), got.GetRefused())
+	// The refusal is the whole point of a TOTAL: once the port frees up, a gauge
+	// would read zero of everything and the burst would be gone.
+	held.Close()
+	waitConns(t, m, "e1", 0)
+	got = counters(t, m, spec, "e1")
+	assert.Zero(t, got.GetActive(), "active is a gauge and comes back down")
+	assert.Equal(t, int64(1), got.GetRefused(), "a refusal already counted is not un-counted")
+}
+// TestExposureCountsACallerItCannotCarry separates the port's own limit from the
+// guest not being there: an operator reading "refused" should raise the cap, and
+// one reading "dropped" should look at the guest.
+func TestExposureCountsACallerItCannotCarry(t *testing.T) {
+	m := newTestManager(t, map[string]string{}) // no address for vm1: still leasing
+	spec := []*pb.ExposureDesired{desired("e1", "vm1", 8080, 0)}
+	m.Converge(spec)
+	addr := boundPort(t, m, "e1")
+	c, err := net.Dial("tcp", addr)
+	require.NoError(t, err)
+	defer c.Close()
+	require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
+	_, err = io.ReadAll(c)
+	require.NoError(t, err)
+	waitConns(t, m, "e1", 0)
+	got := counters(t, m, spec, "e1")
+	assert.Equal(t, int64(1), got.GetDropped(), "a caller the host could not pipe to the guest is a drop")
+	assert.Zero(t, got.GetRefused(), "the cap refused nobody — the guest was not there")
+}
+// TestAFailedExposureStillReportsItsCounters: a port whose bind failed carries
+// the message anyway, so "this agent counts" and "this port has counted
+// nothing" stay one answer apart from "nobody said" for every exposure alike.
+func TestAFailedExposureStillReportsItsCounters(t *testing.T) {
+	held, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer held.Close()
+	m := newTestManager(t, map[string]string{"vm1": "127.0.0.1"})
+	port := portOf(t, held.Addr())
+	out := m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", 8080, port)})
+	require.Len(t, out, 1)
+	require.Equal(t, "failed", out[0].GetState())
+	require.NotNil(t, out[0].GetSessions())
+	assert.Zero(t, out[0].GetSessions().GetActive())
+}

internal/agent/exposeproxy/udp.go

Old		New
@@ -202,12 +202,14 @@ func (m Manager) serve(id string, ex exposure, pc *net.UDPConn, vmID string, g
202		202
203	s := ex.session(client.String())	203	s := ex.session(client.String())
204	if s == nil {	204	if s == nil {
205	s = m.openSession(ex, pc, client, vmID, guestPort, &full)	205	s = m.openSession(id, ex, pc, client, vmID, guestPort, &full)
206	if s == nil {	206	if s == nil {
207	continue	207	continue
208	}	208	}
209	}	209	}
210	if _, err := s.guest.Write(buf[:n]); err != nil {	210	if _, err := s.guest.Write(buf[:n]); err != nil {
		211	ex.dropped.Add(1)
		212	slog.Debug("exposure session torn down, send to guest failed", "exposure", id, "client", s.key, "err", err)
211	ex.dropSession(s)	213	ex.dropSession(s)
212	continue	214	continue
213	}	215	}
@@ -220,22 +222,32 @@ func (m Manager) serve(id string, ex exposure, pc *net.UDPConn, vmID string, g
220	// datagram that asked for it is to be dropped — no guest address, a socket the	222	// datagram that asked for it is to be dropped — no guest address, a socket the
221	// OS would not give, or a table with no room — with full carrying the	223	// OS would not give, or a table with no room — with full carrying the
222	// at-capacity streak so a saturated port logs once rather than per packet.	224	// at-capacity streak so a saturated port logs once rather than per packet.
223	func (m Manager) openSession(ex exposure, pc net.UDPConn, client net.UDPAddr, vmID string, guestPort uint32, full bool) udpSession {	225	//
		226	// The two ways of returning nil are counted apart, because they are different
		227	// answers to an operator's question. A table at its cap is this proxy's own
		228	// limit and says raise it; everything else here is the guest not being
		229	// reachable yet, and says look at the guest.
		230	func (m Manager) openSession(id string, ex exposure, pc net.UDPConn, client net.UDPAddr, vmID string, guestPort uint32, full bool) udpSession {
224	ip := m.addr(vmID)	231	ip := m.addr(vmID)
225	if ip == "" {	232	if ip == "" {
		233	ex.dropped.Add(1)
226	return nil	234	return nil
227	}	235	}
228	guestAddr, err := net.ResolveUDPAddr("udp4", net.JoinHostPort(ip, strconv.Itoa(int(guestPort))))	236	guestAddr, err := net.ResolveUDPAddr("udp4", net.JoinHostPort(ip, strconv.Itoa(int(guestPort))))
229	if err != nil {	237	if err != nil {
		238	ex.dropped.Add(1)
230	return nil	239	return nil
231	}	240	}
232	guest, err := net.DialUDP("udp4", nil, guestAddr)	241	guest, err := net.DialUDP("udp4", nil, guestAddr)
233	if err != nil {	242	if err != nil {
		243	ex.dropped.Add(1)
234	return nil	244	return nil
235	}	245	}
236	s := newUDPSession(client, guest, ip, m.unrepliedIdle, m.repliedIdle)	246	s := newUDPSession(client, guest, ip, m.unrepliedIdle, m.repliedIdle)
237	if !ex.addSession(s, m.maxSessions) {	247	if !ex.addSession(s, m.maxSessions) {
238	s.close()	248	s.close()
		249	ex.refused.Add(1)
		250	slog.Debug("exposure refused a session at its cap", "exposure", id, "cap", m.maxSessions, "client", client.String())
239	if !*full {	251	if !*full {
240	*full = true	252	*full = true
241	slog.Warn("exposure at its UDP session cap, dropping", "cap", m.maxSessions, "client", client.String())	253	slog.Warn("exposure at its UDP session cap, dropping", "cap", m.maxSessions, "client", client.String())
@@ -243,7 +255,8 @@ func (m Manager) openSession(ex exposure, pc net.UDPConn, client net.UDPAddr
243	return nil	255	return nil
244	}	256	}
245	*full = false	257	*full = false
246	go m.relay(ex, pc, s)	258	slog.Debug("exposure session opened", "exposure", id, "client", client.String(), "guest", guestAddr.String())
		259	go m.relay(id, ex, pc, s)
247	return s	260	return s
248	}	261	}
249		262
@@ -257,8 +270,11 @@ func (m Manager) openSession(ex exposure, pc net.UDPConn, client net.UDPAddr
257	// thing that needs to notice an expired session is the goroutine already	270	// thing that needs to notice an expired session is the goroutine already
258	// waiting on it. A deadline that fires early because the forward direction	271	// waiting on it. A deadline that fires early because the forward direction
259	// moved the session on is simply re-armed against the newer expiry.	272	// moved the session on is simply re-armed against the newer expiry.
260	func (m Manager) relay(ex exposure, pc net.UDPConn, s udpSession) {	273	func (m Manager) relay(id string, ex exposure, pc net.UDPConn, s udpSession) {
261	defer ex.dropSession(s)	274	defer func() {
		275	ex.dropSession(s)
		276	slog.Debug("exposure session closed", "exposure", id, "client", s.key)
		277	}()
262	buf := make([]byte, udpDatagramMax)	278	buf := make([]byte, udpDatagramMax)
263	for {	279	for {
264	if err := s.guest.SetReadDeadline(s.expiry()); err != nil {	280	if err := s.guest.SetReadDeadline(s.expiry()); err != nil {

internal/agent/exposeproxy/udp_test.go

@@ @@ -384,3 +384,56 @@ func mustUDPAddr(t *testing.T, addr string) *net.UDPAddr { @@
+	require.NoError(t, err)
+	return ua
+// TestUDPCountsItsSessionsAndRefusals is the counters' UDP half: sessions are
+// what a published UDP port holds, and a full table turns callers away with
+// nothing to show for it on the wire — no connection to refuse, no error to
+// return, just a datagram that goes nowhere.
+func TestUDPCountsItsSessionsAndRefusals(t *testing.T) {
+	g := newFakeUDPGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	m.maxSessions = 1
+	spec := []*pb.ExposureDesired{desiredUDP("e1", "vm1", g.port, 0)}
+	m.Converge(spec)
+	addr := boundUDPPort(t, m, "e1")
+	held := udpClient(t, addr)
+	require.Equal(t, "echo:mine", say(t, held, "mine"))
+	waitSessions(t, m, "e1", 1)
+	got := counters(t, m, spec, "e1")
+	assert.Equal(t, int64(1), got.GetActive(), "a live conversation is one active session")
+	assert.Zero(t, got.GetRefused())
+	over := udpClient(t, addr)
+	_, err := over.Write([]byte("me too"))
+	require.NoError(t, err)
+	require.NoError(t, over.SetReadDeadline(time.Now().Add(300*time.Millisecond)))
+	_, err = over.Read(make([]byte, 64))
+	require.Error(t, err, "precondition: the table is full and the datagram went nowhere")
+	got = counters(t, m, spec, "e1")
+	assert.Equal(t, int64(1), got.GetRefused(), "the dropped datagram is the only trace that caller left")
+	assert.Equal(t, int64(1), got.GetActive(), "and the conversation that was working is untouched")
+}
+// TestUDPCountsADatagramForAGuestWithNoAddress: a guest still leasing is not a
+// port at its cap, and the two must not read the same in a report.
+func TestUDPCountsADatagramForAGuestWithNoAddress(t *testing.T) {
+	m := newTestManager(t, map[string]string{}) // vm1 has no address yet
+	spec := []*pb.ExposureDesired{desiredUDP("e1", "vm1", 9999, 0)}
+	m.Converge(spec)
+	addr := boundUDPPort(t, m, "e1")
+	c := udpClient(t, addr)
+	_, err := c.Write([]byte("anyone there"))
+	require.NoError(t, err)
+	require.NoError(t, c.SetReadDeadline(time.Now().Add(300*time.Millisecond)))
+	_, err = c.Read(make([]byte, 64))
+	require.Error(t, err)
+	got := counters(t, m, spec, "e1")
+	assert.Equal(t, int64(1), got.GetDropped())
+	assert.Zero(t, got.GetRefused())
+	assert.Zero(t, got.GetActive(), "a dropped datagram starts no session")
+}

internal/agent/imagecache/imagecache.go

Old		New
@@ -61,6 +61,11 @@ type Cache struct {
61	// cache over cap; use block-based accounting if that ever bites.	61	// cache over cap; use block-based accounting if that ever bites.
62	MaxBytes int64	62	MaxBytes int64
63		63
		64	// progressStep is the package constant, held per-Cache so a test can watch
		65	// a download advance over a handful of kilobytes instead of gigabytes.
		66	// Written once at construction and only read after.
		67	progressStep int64
		68
64	// fetching collapses concurrent Ensure calls for the same image into one	69	// fetching collapses concurrent Ensure calls for the same image into one
65	// download+decode. Per-VM reconcile workers made creates concurrent, so a	70	// download+decode. Per-VM reconcile workers made creates concurrent, so a
66	// fleet rolling out one image now fetches it from every VM's worker at once;	71	// fleet rolling out one image now fetches it from every VM's worker at once;
@@ -73,12 +78,47 @@ type Cache struct {
73	// so large images on slow links still complete, but a hung connection cannot	78	// so large images on slow links still complete, but a hung connection cannot
74	// stall a reconcile worker forever.	79	// stall a reconcile worker forever.
75	func New(dir string) *Cache {	80	func New(dir string) *Cache {
76	return &Cache{dir: dir, http: &http.Client{Timeout: 10 * time.Minute}}	81	return &Cache{dir: dir, http: &http.Client{Timeout: 10 * time.Minute}, progressStep: defaultProgressStep}
		82	}
		83
		84	// defaultProgressStep is how much has to move before a download says so again.
		85	// A callback per Read would fire thousands of times for one image; every 32 MiB
		86	// is often enough that a watcher sees the number climb and rare enough that
		87	// what it feeds — a report row published under a lock — costs nothing.
		88	const defaultProgressStep = 32 << 20
		89
		90	// progressWriter counts bytes past it and calls report as they pass, at most
		91	// once per step. It writes nothing: it rides the download's existing
		92	// MultiWriter, beside the file and the hash, so nothing in the copy path has to
		93	// know it is there.
		94	//
		95	// total is what the server said the body is, and it is <= 0 whenever the server
		96	// declined to say — a chunked response has no length. Consumers render that as
		97	// "how far", not "how far of what".
		98	type progressWriter struct {
		99	total int64
		100	step int64
		101	done int64
		102	marked int64
		103	report func(done, total int64)
		104	}
		105
		106	func (w *progressWriter) Write(p []byte) (int, error) {
		107	w.done += int64(len(p))
		108	if w.done-w.marked >= w.step {
		109	w.marked = w.done
		110	w.report(w.done, w.total)
		111	}
		112	return len(p), nil
77	}	113	}
78		114
79	// fetch downloads url into a temp file in the cache dir and verifies its	115	// fetch downloads url into a temp file in the cache dir and verifies its
80	// sha256. Returns the temp path; the caller owns renaming or removing it.	116	// sha256. Returns the temp path; the caller owns renaming or removing it.
81	func (c *Cache) fetch(ctx context.Context, url, sha string) (string, error) {	117	//
		118	// progress, when non-nil, is told how far the body has come as it comes. It is
		119	// called from the copy, so it must not block: a slow callback is a slow
		120	// download.
		121	func (c *Cache) fetch(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error) {
82	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)	122	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
83	if err != nil {	123	if err != nil {
84	return "", err	124	return "", err
@@ -96,7 +136,11 @@ func (c *Cache) fetch(ctx context.Context, url, sha string) (string, error) {
96	return "", err	136	return "", err
97	}	137	}
98	h := sha256.New()	138	h := sha256.New()
99	if _, err := io.Copy(io.MultiWriter(tmp, h), resp.Body); err != nil {	139	sink := []io.Writer{tmp, h}
		140	if progress != nil {
		141	sink = append(sink, &progressWriter{total: resp.ContentLength, step: c.progressStep, report: progress})
		142	}
		143	if _, err := io.Copy(io.MultiWriter(sink...), resp.Body); err != nil {
100	tmp.Close()	144	tmp.Close()
101	os.Remove(tmp.Name())	145	os.Remove(tmp.Name())
102	return "", err	146	return "", err
@@ -124,14 +168,21 @@ func (c *Cache) fetch(ctx context.Context, url, sha string) (string, error) {
124	// bounded by the FIRST caller's context, so if that caller's pass times out the	168	// bounded by the FIRST caller's context, so if that caller's pass times out the
125	// waiters inherit its error; and a waiter whose own context expires first stops	169	// waiters inherit its error; and a waiter whose own context expires first stops
126	// waiting without cancelling the fetch, which continues for the others.	170	// waiting without cancelling the fetch, which continues for the others.
127	func (c *Cache) Ensure(ctx context.Context, url, sha string) (string, error) {	171	//
		172	// progress, when non-nil, is told how far the download has come. It belongs to
		173	// the caller that DOES the work: a caller collapsed onto another's fetch is not
		174	// the one reading the body, so its callback never fires and it sees only that
		175	// an image is being fetched. That is the honest reading — the bytes are not
		176	// its download — and it is why the caller's own wording has to stand on its
		177	// own without any numbers in it.
		178	func (c *Cache) Ensure(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error) {
128	// Guard path traversal: sha becomes part of the cache file path. Checked	179	// Guard path traversal: sha becomes part of the cache file path. Checked
129	// before the singleflight so a bad digest can never key an entry.	180	// before the singleflight so a bad digest can never key an entry.
130	if !sha256Re.MatchString(sha) {	181	if !sha256Re.MatchString(sha) {
131	return "", fmt.Errorf("invalid sha256: %q", sha)	182	return "", fmt.Errorf("invalid sha256: %q", sha)
132	}	183	}
133		184
134	ch := c.fetching.DoChan(sha, func() (any, error) { return c.ensureOnce(ctx, url, sha) })	185	ch := c.fetching.DoChan(sha, func() (any, error) { return c.ensureOnce(ctx, url, sha, progress) })
135	select {	186	select {
136	case r := <-ch:	187	case r := <-ch:
137	if r.Err != nil {	188	if r.Err != nil {
@@ -145,7 +196,7 @@ func (c *Cache) Ensure(ctx context.Context, url, sha string) (string, error) {
145		196
146	// ensureOnce is the un-deduplicated body of Ensure: at most one runs per sha at	197	// ensureOnce is the un-deduplicated body of Ensure: at most one runs per sha at
147	// a time.	198	// a time.
148	func (c *Cache) ensureOnce(ctx context.Context, url, sha string) (string, error) {	199	func (c *Cache) ensureOnce(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error) {
149	final := filepath.Join(c.dir, sha+".raw")	200	final := filepath.Join(c.dir, sha+".raw")
150	if _, err := os.Stat(final); err == nil {	201	if _, err := os.Stat(final); err == nil {
151	// Hit: refresh recency so frequently-used images sort as recent.	202	// Hit: refresh recency so frequently-used images sort as recent.
@@ -154,7 +205,7 @@ func (c *Cache) ensureOnce(ctx context.Context, url, sha string) (string, error)
154	c.evict(final)	205	c.evict(final)
155	return final, nil	206	return final, nil
156	}	207	}
157	tmp, err := c.fetch(ctx, url, sha)	208	tmp, err := c.fetch(ctx, url, sha, progress)
158	if err != nil {	209	if err != nil {
159	return "", err	210	return "", err
160	}	211	}

internal/agent/imagecache/imagecache_test.go

Old		New
@@ -10,6 +10,7 @@ import (
10	"net/http/httptest"	10	"net/http/httptest"
11	"os"	11	"os"
12	"path/filepath"	12	"path/filepath"
		13	"strconv"
13	"sync/atomic"	14	"sync/atomic"
14	"testing"	15	"testing"
15	"time"	16	"time"
@@ -65,11 +66,11 @@ func TestEnsureDownloadsVerifiesAndCaches(t *testing.T) {
65	sha := hex.EncodeToString(sum[:])	66	sha := hex.EncodeToString(sum[:])
66	c := New(t.TempDir())	67	c := New(t.TempDir())
67		68
68	p1, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sha)	69	p1, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sha, nil)
69	require.NoError(t, err)	70	require.NoError(t, err)
70	assert.FileExists(t, p1)	71	assert.FileExists(t, p1)
71		72
72	p2, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sha)	73	p2, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sha, nil)
73	require.NoError(t, err)	74	require.NoError(t, err)
74	assert.Equal(t, p1, p2)	75	assert.Equal(t, p1, p2)
75	assert.Equal(t, int64(1), downloads.Load(), "second Ensure must hit the cache, not re-fetch")	76	assert.Equal(t, int64(1), downloads.Load(), "second Ensure must hit the cache, not re-fetch")
@@ -83,7 +84,7 @@ func TestEnsureMaterializesRawByRename(t *testing.T) {
83	ts, sum := serve(t, body)	84	ts, sum := serve(t, body)
84	c := New(t.TempDir())	85	c := New(t.TempDir())
85		86
86	p, err := c.Ensure(context.Background(), ts.URL+"/disk.raw", sum)	87	p, err := c.Ensure(context.Background(), ts.URL+"/disk.raw", sum, nil)
87	require.NoError(t, err)	88	require.NoError(t, err)
88		89
89	got, err := os.ReadFile(p)	90	got, err := os.ReadFile(p)
@@ -99,7 +100,7 @@ func TestEnsureDecodesQcow2Natively(t *testing.T) {
99	ts, sum := serve(t, tinyQcow2(t))	100	ts, sum := serve(t, tinyQcow2(t))
100	c := New(t.TempDir())	101	c := New(t.TempDir())
101		102
102	p, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum)	103	p, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum, nil)
103	require.NoError(t, err)	104	require.NoError(t, err)
104		105
105	got, err := os.ReadFile(p)	106	got, err := os.ReadFile(p)
@@ -120,7 +121,7 @@ func TestEnsureRejectsUnparsableImagePermanently(t *testing.T) {
120	dir := t.TempDir()	121	dir := t.TempDir()
121	c := New(dir)	122	c := New(dir)
122		123
123	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum)	124	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum, nil)
124	require.Error(t, err)	125	require.Error(t, err)
125	var perm interface{ Permanent() bool }	126	var perm interface{ Permanent() bool }
126	require.ErrorAs(t, err, &perm, "a broken container header must be marked permanent")	127	require.ErrorAs(t, err, &perm, "a broken container header must be marked permanent")
@@ -134,7 +135,7 @@ func TestEnsureRejectsUnparsableImagePermanently(t *testing.T) {
134	func TestEnsureRejectsChecksumMismatch(t *testing.T) {	135	func TestEnsureRejectsChecksumMismatch(t *testing.T) {
135	ts, _ := serve(t, []byte("evil-bytes"))	136	ts, _ := serve(t, []byte("evil-bytes"))
136	c := New(t.TempDir())	137	c := New(t.TempDir())
137	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", "0000000000000000000000000000000000000000000000000000000000000000")	138	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", "0000000000000000000000000000000000000000000000000000000000000000", nil)
138	assert.Error(t, err, "tampered image must be rejected before conversion")	139	assert.Error(t, err, "tampered image must be rejected before conversion")
139	}	140	}
140		141
@@ -144,19 +145,19 @@ func TestEnsureRejectsInvalidSha(t *testing.T) {
144	c := New(t.TempDir())	145	c := New(t.TempDir())
145		146
146	t.Run("path traversal", func(t *testing.T) {	147	t.Run("path traversal", func(t *testing.T) {
147	_, err := c.Ensure(context.Background(), "http://unused", "../../etc/passwd")	148	_, err := c.Ensure(context.Background(), "http://unused", "../../etc/passwd", nil)
148	assert.Error(t, err, "path traversal sha must be rejected")	149	assert.Error(t, err, "path traversal sha must be rejected")
149	assert.Contains(t, err.Error(), "invalid sha256")	150	assert.Contains(t, err.Error(), "invalid sha256")
150	})	151	})
151		152
152	t.Run("uppercase hex", func(t *testing.T) {	153	t.Run("uppercase hex", func(t *testing.T) {
153	_, err := c.Ensure(context.Background(), "http://unused",	154	_, err := c.Ensure(context.Background(), "http://unused",
154	"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")	155	"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", nil)
155	assert.Error(t, err, "uppercase sha must be rejected")	156	assert.Error(t, err, "uppercase sha must be rejected")
156	})	157	})
157		158
158	t.Run("too short", func(t *testing.T) {	159	t.Run("too short", func(t *testing.T) {
159	_, err := c.Ensure(context.Background(), "http://unused", "abc123")	160	_, err := c.Ensure(context.Background(), "http://unused", "abc123", nil)
160	assert.Error(t, err)	161	assert.Error(t, err)
161	})	162	})
162	}	163	}
@@ -176,7 +177,7 @@ func TestEnsureAtomicConvert_LeftoverPartialIsIgnored(t *testing.T) {
176	require.NoError(t, os.WriteFile(partial, []byte("corrupt partial"), 0o644))	177	require.NoError(t, os.WriteFile(partial, []byte("corrupt partial"), 0o644))
177		178
178	c := New(dir)	179	c := New(dir)
179	p, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum)	180	p, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum, nil)
180	require.NoError(t, err)	181	require.NoError(t, err)
181		182
182	// The result must contain the decoded image, not the corrupt partial.	183	// The result must contain the decoded image, not the corrupt partial.
@@ -191,7 +192,7 @@ func TestEnsureNoStrayFilesAfterSuccess(t *testing.T) {
191	dir := t.TempDir()	192	dir := t.TempDir()
192	c := New(dir)	193	c := New(dir)
193		194
194	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum)	195	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2", sum, nil)
195	require.NoError(t, err)	196	require.NoError(t, err)
196		197
197	entries, err := os.ReadDir(dir)	198	entries, err := os.ReadDir(dir)
@@ -204,7 +205,7 @@ func TestEnsureNoStrayFilesAfterSuccess(t *testing.T) {
204	func ensureBytes(t testing.T, c Cache, body []byte) string {	205	func ensureBytes(t testing.T, c Cache, body []byte) string {
205	t.Helper()	206	t.Helper()
206	ts, sha := serve(t, body)	207	ts, sha := serve(t, body)
207	p, err := c.Ensure(context.Background(), ts.URL+"/img", sha)	208	p, err := c.Ensure(context.Background(), ts.URL+"/img", sha, nil)
208	require.NoError(t, err)	209	require.NoError(t, err)
209	return p	210	return p
210	}	211	}
@@ -325,7 +326,7 @@ func TestConcurrentEnsureFetchesOnce(t *testing.T) {
325	errs := make(chan error, callers)	326	errs := make(chan error, callers)
326	for range callers {	327	for range callers {
327	go func() {	328	go func() {
328	p, err := c.Ensure(context.Background(), srv.URL, sha)	329	p, err := c.Ensure(context.Background(), srv.URL, sha, nil)
329	if err != nil {	330	if err != nil {
330	errs <- err	331	errs <- err
331	return	332	return
@@ -389,7 +390,7 @@ func TestEnsureDecompressesGzippedRaw(t *testing.T) {
389	dir := t.TempDir()	390	dir := t.TempDir()
390	c := New(dir)	391	c := New(dir)
391		392
392	p, err := c.Ensure(context.Background(), ts.URL+"/img.raw.gz", sum)	393	p, err := c.Ensure(context.Background(), ts.URL+"/img.raw.gz", sum, nil)
393	require.NoError(t, err)	394	require.NoError(t, err)
394		395
395	got, err := os.ReadFile(p)	396	got, err := os.ReadFile(p)
@@ -412,7 +413,7 @@ func TestEnsureRejectsCorruptGzipPermanently(t *testing.T) {
412	ts, sum := serve(t, []byte{0x1f, 0x8b, 'g', 'a', 'r', 'b', 'a', 'g', 'e', '!'})	413	ts, sum := serve(t, []byte{0x1f, 0x8b, 'g', 'a', 'r', 'b', 'a', 'g', 'e', '!'})
413	c := New(t.TempDir())	414	c := New(t.TempDir())
414		415
415	_, err := c.Ensure(context.Background(), ts.URL+"/img.raw.gz", sum)	416	_, err := c.Ensure(context.Background(), ts.URL+"/img.raw.gz", sum, nil)
416	require.Error(t, err)	417	require.Error(t, err)
417	var p interface{ Permanent() bool }	418	var p interface{ Permanent() bool }
418	require.ErrorAs(t, err, &p, "a broken gzip header must be marked permanent")	419	require.ErrorAs(t, err, &p, "a broken gzip header must be marked permanent")
@@ -429,7 +430,7 @@ func TestEnsureRejectsGzippedQcow2Permanently(t *testing.T) {
429	dir := t.TempDir()	430	dir := t.TempDir()
430	c := New(dir)	431	c := New(dir)
431		432
432	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2.gz", sum)	433	_, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2.gz", sum, nil)
433	require.Error(t, err)	434	require.Error(t, err)
434	var p interface{ Permanent() bool }	435	var p interface{ Permanent() bool }
435	require.ErrorAs(t, err, &p, "a gzipped qcow2 can never decompress to a raw image")	436	require.ErrorAs(t, err, &p, "a gzipped qcow2 can never decompress to a raw image")
@@ -440,3 +441,86 @@ func TestEnsureRejectsGzippedQcow2Permanently(t *testing.T) {
440	require.NoError(t, err)	441	require.NoError(t, err)
441	assert.Empty(t, entries, "a rejected image must leave nothing in the cache")	442	assert.Empty(t, entries, "a rejected image must leave nothing in the cache")
442	}	443	}
		444
		445	// TestEnsureReportsDownloadProgress is the narration leg: a multi-gigabyte
		446	// image is minutes of silence unless the download says how far it has come, and
		447	// the length the server declared is what turns "how far" into "how far of what".
		448	func TestEnsureReportsDownloadProgress(t *testing.T) {
		449	body := bytes.Repeat([]byte{0x00, 0xeb, 0x63, 0x90}, 4096) // 16 KiB of raw
		450	// Declares its length, the way a server handing out an image file does.
		451	// httptest's default is chunked, which is the other case entirely.
		452	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		453	w.Header().Set("Content-Length", strconv.Itoa(len(body)))
		454	w.Write(body)
		455	}))
		456	t.Cleanup(ts.Close)
		457	digest := sha256.Sum256(body)
		458	sum := hex.EncodeToString(digest[:])
		459	c := New(t.TempDir())
		460	c.progressStep = 4 << 10
		461
		462	type step struct{ done, total int64 }
		463	var steps []step
		464	_, err := c.Ensure(context.Background(), ts.URL+"/disk.raw", sum, func(done, total int64) {
		465	steps = append(steps, step{done, total})
		466	})
		467	require.NoError(t, err)
		468
		469	require.NotEmpty(t, steps, "a download must say how far it has come")
		470	for _, s := range steps {
		471	assert.Equal(t, int64(len(body)), s.total, "the declared length rides every report")
		472	assert.LessOrEqual(t, s.done, int64(len(body)))
		473	}
		474	for i := 1; i < len(steps); i++ {
		475	assert.Greater(t, steps[i].done, steps[i-1].done, "progress only moves forward")
		476	}
		477	}
		478
		479	// TestEnsureProgressWithoutDeclaredLength: a server that will not say how long
		480	// the body is (chunked) still gets counted, with a total of -1 saying so. The
		481	// renderer downstream needs that difference to be legible, not guessed at.
		482	func TestEnsureProgressWithoutDeclaredLength(t *testing.T) {
		483	body := bytes.Repeat([]byte{0x00, 0xeb, 0x63, 0x90}, 4096)
		484	ts, sum := serve(t, body) // httptest without a Content-Length: chunked
		485	c := New(t.TempDir())
		486	c.progressStep = 4 << 10
		487
		488	var totals []int64
		489	_, err := c.Ensure(context.Background(), ts.URL+"/disk.raw", sum, func(_, total int64) {
		490	totals = append(totals, total)
		491	})
		492	require.NoError(t, err)
		493	require.NotEmpty(t, totals)
		494	for _, total := range totals {
		495	assert.Negative(t, total, "an undeclared length is reported as unknown, not as zero")
		496	}
		497	}
		498
		499	// TestProgressThrottlesToItsStep pins the throttle itself: the callback feeds a
		500	// published report row, so a download must not call it once per read.
		501	func TestProgressThrottlesToItsStep(t *testing.T) {
		502	var calls int
		503	w := &progressWriter{total: 400, step: 100, report: func(int64, int64) { calls++ }}
		504	for range 40 {
		505	_, err := w.Write(make([]byte, 10))
		506	require.NoError(t, err)
		507	}
		508	assert.Equal(t, 4, calls, "40 writes of 10 bytes past a 100-byte step is 4 reports, not 40")
		509	}
		510
		511	// TestEnsureWithoutProgressStaysQuiet: the callback is optional, and a cache hit
		512	// downloads nothing to report on.
		513	func TestEnsureCacheHitReportsNothing(t *testing.T) {
		514	body := bytes.Repeat([]byte{0x00, 0xeb, 0x63, 0x90}, 1024)
		515	ts, sum := serve(t, body)
		516	c := New(t.TempDir())
		517	c.progressStep = 1
		518
		519	_, err := c.Ensure(context.Background(), ts.URL+"/disk.raw", sum, nil)
		520	require.NoError(t, err)
		521
		522	called := false
		523	_, err = c.Ensure(context.Background(), ts.URL+"/disk.raw", sum, func(int64, int64) { called = true })
		524	require.NoError(t, err)
		525	assert.False(t, called, "a cache hit fetches nothing, so it reports nothing")
		526	}

internal/agent/reconcile/narrate_test.go

@@ @@ -0,0 +1,90 @@ @@
+package reconcile
+import (
+	"context"
+	"testing"
+	"github.com/a73x/eitri/internal/agent/state"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// narrateCreate runs one VM's create with a publisher attached and returns
+// every sentence it published, in order. This is the seam the worker uses in
+// production (see worker.publish) — a pass that has not finished still has
+// something to say.
+func narrateCreate(t *testing.T, f *fixture, id string) []string {
+	t.Helper()
+	var said []string
+	var res vmResult
+	f.eng.create(context.Background(), vm(id), state.Record{}, false, &res,
+		func(interim vmResult) {
+			require.Equal(t, "creating", interim.vm.GetPhase(), "narration never changes the phase")
+			said = append(said, interim.vm.GetStatusDetail())
+		})
+	return said
+}
+// TestACreateSaysWhatItIsDoing is the whole point of the field: `creating` is
+// one word for image, disk, seed and boot, and only the host can see which of
+// them a VM is sitting in.
+func TestACreateSaysWhatItIsDoing(t *testing.T) {
+	f := setup(t)
+	said := narrateCreate(t, f, "vm1")
+	assert.Equal(t, []string{
+		"waiting for a create slot on this host",
+		"downloading image",
+		"preparing root disk",
+		"building the cloud-init seed",
+		"booting",
+	}, said)
+}
+// TestACreateCountsOutTheDownload pins the one step that takes long enough to
+// need a number rather than a name.
+func TestACreateCountsOutTheDownload(t *testing.T) {
+	f := setup(t)
+	f.eng.Images = func(_ context.Context, _, sha string, progress func(done, total int64)) (string, error) {
+		progress(1_288_490_188, 3_972_844_748) // 1.2 of 3.7 GiB
+		return "/cache/" + sha + ".raw", nil
+	}
+	assert.Contains(t, narrateCreate(t, f, "vm1"), "downloading image 1.2/3.7 GiB")
+}
+// TestNarrationCarriesTheGuestHostKey: the public host key is level-triggered —
+// it must be in EVERY report for as long as the VM exists, or the control plane
+// stops re-certifying it. A row published mid-pass is a report row like any
+// other and cannot be the one that drops it.
+func TestNarrationCarriesTheGuestHostKey(t *testing.T) {
+	f := setup(t)
+	var res vmResult
+	res.hostPubKey = "ssh-ed25519 AAAA-test guest"
+	published := 0
+	f.eng.create(context.Background(), vm("vm1"), state.Record{}, false, &res,
+		func(interim vmResult) {
+			published++
+			assert.Equal(t, res.hostPubKey, interim.hostPubKey)
+		})
+	assert.Positive(t, published)
+}
+// TestACreateWithNobodyListeningStillCreates: narration is commentary, and a
+// pass with no publisher must behave exactly as it always did.
+func TestACreateWithNobodyListeningStillCreates(t *testing.T) {
+	f := setup(t)
+	var res vmResult
+	f.eng.create(context.Background(), vm("vm1"), state.Record{}, false, &res, nil)
+	require.NotNil(t, res.vm)
+	assert.Equal(t, "ready", res.vm.GetPhase())
+	assert.Empty(t, res.vm.GetStatusDetail(), "a settled VM has nothing to add")
+}
+func TestDownloadDetailWords(t *testing.T) {
+	assert.Equal(t, "downloading image 1.2/3.7 GiB", downloadDetail(1_288_490_188, 3_972_844_748))
+	// Both halves scale together off the larger, so the pair reads as a fraction
+	// rather than as arithmetic between two units.
+	assert.Equal(t, "downloading image 0.4/3.7 GiB", downloadDetail(429_496_730, 3_972_844_748))
+	assert.Equal(t, "downloading image 120.0/512.0 MiB", downloadDetail(125_829_120, 536_870_912))
+	// A server that declared no length gets the half it can vouch for.
+	assert.Equal(t, "downloading image 1.2 GiB", downloadDetail(1_288_490_188, -1))
+}

internal/agent/reconcile/reconcile.go

Old		New
@@ -118,7 +118,13 @@ type Engine struct {
118		118
119	// Images resolves an image URL+sha256 to a local base-image path, fetching	119	// Images resolves an image URL+sha256 to a local base-image path, fetching
120	// if necessary. Returns the path to the raw base image.	120	// if necessary. Returns the path to the raw base image.
121	Images func(ctx context.Context, url, sha string) (string, error)	121	//
		122	// progress is called as the download advances, and is the one thing on this
		123	// seam that exists for the operator rather than for the VM: an image is
		124	// gigabytes and a create is otherwise one word for the whole time it takes.
		125	// An implementation that cannot say (a cache hit, a fetch collapsed onto
		126	// another VM's) simply never calls it.
		127	Images func(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error)
122		128
123	// Seed builds the cloud-init NoCloud seed ISO at outPath.	129	// Seed builds the cloud-init NoCloud seed ISO at outPath.
124	Seed func(outPath string, p seed.Params) error	130	Seed func(outPath string, p seed.Params) error
@@ -349,6 +355,17 @@ func (e Engine) fenceReport(currentEpoch uint64) pb.ActualStateReport {
349	return rep	355	return rep
350	}	356	}
351		357
		358	// publisher hands a VM's row upward mid-pass, before the pass that owns it has
		359	// finished. A pass publishes its result when it ENDS, and a first create ends
		360	// minutes after it starts — image, disk, seed, boot — so the slowest and most
		361	// opaque part of a VM's life was the part the host report carried no row for at
		362	// all, and everything above it could say was "creating".
		363	//
		364	// Nil is a caller with nothing collecting rows (the fence path, a test calling
		365	// a pass directly), and narration is then skipped rather than being an error:
		366	// what it produces is commentary, and no decision anywhere rests on it.
		367	type publisher func(vmResult)
		368
352	// reconcileOne is ONE VM's complete reconcile pass: bounded by VMTimeout, it	369	// reconcileOne is ONE VM's complete reconcile pass: bounded by VMTimeout, it
353	// loads that VM's own record and either drives it toward desired or reaps it.	370	// loads that VM's own record and either drives it toward desired or reaps it.
354	// It is the entire unit of work a per-VM worker runs, and it touches no other	371	// It is the entire unit of work a per-VM worker runs, and it touches no other
@@ -364,7 +381,7 @@ func (e Engine) fenceReport(currentEpoch uint64) pb.ActualStateReport {
364	// store recovers, and the VM keeps its last-known row in the report meanwhile	381	// store recovers, and the VM keeps its last-known row in the report meanwhile
365	// (see worker.run). Acting on the unreadable record instead would route a	382	// (see worker.run). Acting on the unreadable record instead would route a
366	// running VM into create().	383	// running VM into create().
367	func (e *Engine) reconcileOne(ctx context.Context, id string, a assignment) (vmResult, bool) {	384	func (e *Engine) reconcileOne(ctx context.Context, id string, a assignment, pub publisher) (vmResult, bool) {
368	if e.VMTimeout > 0 {	385	if e.VMTimeout > 0 {
369	var cancel context.CancelFunc	386	var cancel context.CancelFunc
370	ctx, cancel = context.WithTimeout(ctx, e.VMTimeout)	387	ctx, cancel = context.WithTimeout(ctx, e.VMTimeout)
@@ -398,7 +415,7 @@ func (e *Engine) reconcileOne(ctx context.Context, id string, a assignment) (vmR
398	rec.QuarantineTombstoned = false	415	rec.QuarantineTombstoned = false
399	_ = e.St.SaveVM(rec)	416	_ = e.St.SaveVM(rec)
400	}	417	}
401	e.reconcileVM(ctx, a.desired, rec, ok, &res)	418	e.reconcileVM(ctx, a.desired, rec, ok, &res, pub)
402	return res, true	419	return res, true
403	}	420	}
404		421
@@ -493,9 +510,9 @@ func (e *Engine) reapVM(ctx context.Context, id string, rec state.Record, isTomb
493	// PrepareRootDisk but before boot leaves a disk on a still-empty BootID. Treating	510	// PrepareRootDisk but before boot leaves a disk on a still-empty BootID. Treating
494	// that disk as "exists" would divert the retry to converge(), which never	511	// that disk as "exists" would divert the retry to converge(), which never
495	// rebuilds the disk or seed, and the VM would never recover.	512	// rebuilds the disk or seed, and the VM would never recover.
496	func (e Engine) reconcileVM(ctx context.Context, d pb.VMDesired, rec state.Record, ok bool, res *vmResult) {	513	func (e Engine) reconcileVM(ctx context.Context, d pb.VMDesired, rec state.Record, ok bool, res *vmResult, pub publisher) {
497	if !ok \|\| rec.BootID == "" {	514	if !ok \|\| rec.BootID == "" {
498	e.create(ctx, d, rec, ok, res)	515	e.create(ctx, d, rec, ok, res, pub)
499	return	516	return
500	}	517	}
501	e.converge(ctx, d, rec, res)	518	e.converge(ctx, d, rec, res)
@@ -635,7 +652,7 @@ func (e *Engine) quotaCheckLocked(vmID string, spec state.VMSpec) string {
635	// prior record (retry budget, last known address) and ok reports whether one	652	// prior record (retry budget, last known address) and ok reports whether one
636	// exists. Quota comes from the serialized admission ledger (see admit); the	653	// exists. Quota comes from the serialized admission ledger (see admit); the
637	// address comes from the backend, at boot.	654	// address comes from the backend, at boot.
638	func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record, ok bool, res *vmResult) {	655	func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record, ok bool, res *vmResult, pub publisher) {
639	// Defensive: never start an attempt (which would burn retry budget) on a	656	// Defensive: never start an attempt (which would burn retry budget) on a
640	// context that is already dead. UNREACHABLE today — a pass context is a	657	// context that is already dead. UNREACHABLE today — a pass context is a
641	// fresh context.Background plus VMTimeout (see worker.run), so it cannot	658	// fresh context.Background plus VMTimeout (see worker.run), so it cannot
@@ -716,6 +733,23 @@ func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record,
716	return	733	return
717	}	734	}
718		735
		736	// say publishes what this create is doing right now, on the row this VM
		737	// would report anyway: still creating, still stopped, no error. It carries
		738	// rec's address and this VM's public host key because BOTH are
		739	// level-triggered — a row that dropped the key would ask the control plane
		740	// to forget a guest's certificate for a tick — and rec is read at each call,
		741	// so a step after Boot narrates with the address Boot found.
		742	say := func(detail string) {
		743	if pub == nil {
		744	return
		745	}
		746	var interim vmResult
		747	interim.hostPubKey = res.hostPubKey
		748	interim.report(d.VmId, rec.IP, "stopped", "creating", "")
		749	interim.vm.StatusDetail = detail
		750	pub(interim)
		751	}
		752
719	// Ask the backend whether this host can run a guest before spending anything	753	// Ask the backend whether this host can run a guest before spending anything
720	// on finding out. Ahead of the throttle as well as the fetch: a create that	754	// on finding out. Ahead of the throttle as well as the fetch: a create that
721	// can only be refused should not queue for a slot that a create which might	755	// can only be refused should not queue for a slot that a create which might
@@ -735,6 +769,7 @@ func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record,
735	// A wait that outlives VMTimeout fails the attempt, and failCreate REFUNDS a	769	// A wait that outlives VMTimeout fails the attempt, and failCreate REFUNDS a
736	// ctx-expiry failure, so a queued VM never burns retry budget for waiting.	770	// ctx-expiry failure, so a queued VM never burns retry budget for waiting.
737	// Held through Boot, which is cheap but immediately I/O-heavy in the guest.	771	// Held through Boot, which is cheap but immediately I/O-heavy in the guest.
		772	say("waiting for a create slot on this host")
738	release, err := e.acquireCreateSlot(ctx)	773	release, err := e.acquireCreateSlot(ctx)
739	if err != nil {	774	if err != nil {
740	e.failCreate(ctx, rec, err, res)	775	e.failCreate(ctx, rec, err, res)
@@ -742,14 +777,21 @@ func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record,
742	}	777	}
743	defer release()	778	defer release()
744		779
745	// Resolve base image.	780	// Resolve base image. The download is the longest thing a create does and
746	basePath, err := e.Images(ctx, d.ImageUrl, d.ImageSha256)	781	// the one an operator most often wants a number for, so it narrates itself
		782	// as it goes; a cached image passes straight through and the line stands for
		783	// only as long as the check takes.
		784	say("downloading image")
		785	basePath, err := e.Images(ctx, d.ImageUrl, d.ImageSha256, func(done, total int64) {
		786	say(downloadDetail(done, total))
		787	})
747	if err != nil {	788	if err != nil {
748	e.failCreate(ctx, rec, err, res)	789	e.failCreate(ctx, rec, err, res)
749	return	790	return
750	}	791	}
751		792
752	// Prepare root disk.	793	// Prepare root disk.
		794	say("preparing root disk")
753	if err := e.Prov.PrepareRootDisk(ctx, rec.Spec, basePath); err != nil {	795	if err := e.Prov.PrepareRootDisk(ctx, rec.Spec, basePath); err != nil {
754	e.failCreate(ctx, rec, err, res)	796	e.failCreate(ctx, rec, err, res)
755	return	797	return
@@ -757,6 +799,7 @@ func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record,
757		799
758	// Build cloud-init seed ISO. The guest gets its address from the host	800	// Build cloud-init seed ISO. The guest gets its address from the host
759	// network, so no IP or gateway is baked into the seed.	801	// network, so no IP or gateway is baked into the seed.
		802	say("building the cloud-init seed")
760	if err := e.Seed(e.St.SeedPath(d.VmId), seed.Params{	803	if err := e.Seed(e.St.SeedPath(d.VmId), seed.Params{
761	Hostname: d.Name,	804	Hostname: d.Name,
762	InstanceID: d.VmId,	805	InstanceID: d.VmId,
@@ -778,6 +821,7 @@ func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record,
778	// report row carrying the address, as it always has — and one that learns	821	// report row carrying the address, as it always has — and one that learns
779	// it later fills it in on a converge poll.	822	// it later fills it in on a converge poll.
780	if d.PowerState == "running" {	823	if d.PowerState == "running" {
		824	say("booting")
781	if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {	825	if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {
782	e.failCreate(ctx, rec, err, res)	826	e.failCreate(ctx, rec, err, res)
783	return	827	return
@@ -1002,6 +1046,31 @@ func (r vmResult) merge(rep pb.ActualStateReport) {
1002	}	1046	}
1003	}	1047	}
1004		1048
		1049	// downloadDetail words how far an image download has come.
		1050	//
		1051	// Both numbers are scaled by the SAME unit, chosen from the larger of them, so
		1052	// the pair can be read against each other at a glance — "0.4/3.7 GiB" is a
		1053	// fraction; "419.4 MiB/3.7 GiB" is arithmetic. A server that declared no length
		1054	// (chunked, total <= 0) gets the only half it can vouch for: how far, with
		1055	// nothing said about how far there is to go.
		1056	func downloadDetail(done, total int64) string {
		1057	scale, unit := byteScale(max(done, total))
		1058	if total <= 0 {
		1059	return fmt.Sprintf("downloading image %.1f %s", float64(done)/scale, unit)
		1060	}
		1061	return fmt.Sprintf("downloading image %.1f/%.1f %s", float64(done)/scale, float64(total)/scale, unit)
		1062	}
		1063
		1064	// byteScale picks the unit a size reads best in: the largest one it is at least
		1065	// one of, floored at MiB because an image smaller than that does not exist and
		1066	// a download reported in kilobytes would look like a failure.
		1067	func byteScale(n int64) (float64, string) {
		1068	if n >= 1<<30 {
		1069	return 1 << 30, "GiB"
		1070	}
		1071	return 1 << 20, "MiB"
		1072	}
		1073
1005	// newActualVM builds one ActualVM row from what a reconcile pass observed.	1074	// newActualVM builds one ActualVM row from what a reconcile pass observed.
1006	// The row's remaining field, ssh_host_pubkey, is stamped by merge — see	1075	// The row's remaining field, ssh_host_pubkey, is stamped by merge — see
1007	// vmResult.hostPubKey. Unset values are the proto zero-value "".	1076	// vmResult.hostPubKey. Unset values are the proto zero-value "".

internal/agent/reconcile/reconcile_test.go

Old		New
@@ -199,7 +199,7 @@ func (f fixture) newEngine() Engine {
199	return &Engine{	199	return &Engine{
200	St: f.st,	200	St: f.st,
201	Prov: f.prov,	201	Prov: f.prov,
202	Images: func(ctx context.Context, url, sha string) (string, error) {	202	Images: func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
203	return "/cache/" + sha + ".raw", nil	203	return "/cache/" + sha + ".raw", nil
204	},	204	},
205	Seed: func(out string, p seed.Params) error { return nil },	205	Seed: func(out string, p seed.Params) error { return nil },
@@ -615,7 +615,7 @@ func TestFenceReportIncludesQuarantinedVMs(t *testing.T) {
615	func TestVMTimeoutBoundsSlowOperations(t *testing.T) {	615	func TestVMTimeoutBoundsSlowOperations(t *testing.T) {
616	f := setup(t)	616	f := setup(t)
617	f.eng.VMTimeout = 50 * time.Millisecond	617	f.eng.VMTimeout = 50 * time.Millisecond
618	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {	618	f.eng.Images = func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
619	<-ctx.Done() // wedged until the watchdog fires	619	<-ctx.Done() // wedged until the watchdog fires
620	return "", ctx.Err()	620	return "", ctx.Err()
621	}	621	}
@@ -639,7 +639,7 @@ func TestVMTimeoutBoundsSlowOperations(t *testing.T) {
639	func TestVMTimeoutZeroDisablesWatchdog(t *testing.T) {	639	func TestVMTimeoutZeroDisablesWatchdog(t *testing.T) {
640	f := setup(t)	640	f := setup(t)
641	sawDeadline := false	641	sawDeadline := false
642	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {	642	f.eng.Images = func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
643	_, sawDeadline = ctx.Deadline()	643	_, sawDeadline = ctx.Deadline()
644	return "/cache/x.raw", nil	644	return "/cache/x.raw", nil
645	}	645	}
@@ -655,7 +655,7 @@ func TestVMTimeoutZeroDisablesWatchdog(t *testing.T) {
655	func TestWatchdogExpiryDoesNotBurnCreateAttempts(t *testing.T) {	655	func TestWatchdogExpiryDoesNotBurnCreateAttempts(t *testing.T) {
656	f := setup(t)	656	f := setup(t)
657	f.eng.VMTimeout = 50 * time.Millisecond	657	f.eng.VMTimeout = 50 * time.Millisecond
658	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {	658	f.eng.Images = func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
659	<-ctx.Done() // every fetch wedges until the watchdog fires	659	<-ctx.Done() // every fetch wedges until the watchdog fires
660	return "", ctx.Err()	660	return "", ctx.Err()
661	}	661	}
@@ -707,7 +707,7 @@ func TestPermanentCreateErrorFailsTerminallyInOneAttempt(t *testing.T) {
707	func TestPreflightRefusalFailsBeforeTheImageFetch(t *testing.T) {	707	func TestPreflightRefusalFailsBeforeTheImageFetch(t *testing.T) {
708	f := setup(t)	708	f := setup(t)
709	var fetches int	709	var fetches int
710	f.eng.Images = func(context.Context, string, string) (string, error) {	710	f.eng.Images = func(context.Context, string, string, func(int64, int64)) (string, error) {
711	fetches++	711	fetches++
712	return "/cache/x.raw", nil	712	return "/cache/x.raw", nil
713	}	713	}

internal/agent/reconcile/worker.go

Old		New
@@ -200,7 +200,7 @@ func (w *worker) run() {
200		200
201	// LOAD-BEARING: the pass runs with w.mu RELEASED. That is what lets	201	// LOAD-BEARING: the pass runs with w.mu RELEASED. That is what lets
202	// deliver and collect run at full speed while this VM does slow work.	202	// deliver and collect run at full speed while this VM does slow work.
203	res, ok := w.eng.reconcileOne(context.Background(), w.id, a)	203	res, ok := w.eng.reconcileOne(context.Background(), w.id, a, w.publish)
204		204
205	w.mu.Lock()	205	w.mu.Lock()
206	// Publish only a pass that ran. A skipped pass (this VM's record could	206	// Publish only a pass that ran. A skipped pass (this VM's record could
@@ -218,6 +218,22 @@ func (w *worker) run() {
218	}	218	}
219	}	219	}
220		220
		221	// publish swaps in a result from INSIDE a running pass, so the host report can
		222	// carry a row for work that has not finished. Every other publish happens at
		223	// the end of a pass (see run); this is the one that happens during one, and it
		224	// exists because the pass that most needs reporting on is the one that takes
		225	// minutes — a first create.
		226	//
		227	// It takes the same lock the pass-end publish takes and holds it no longer, so
		228	// a narrating create is no more able to delay a report than a silent one. The
		229	// pass's own result lands on top when it ends, which is what keeps this
		230	// commentary and not state: nothing here can outlive the pass that said it.
		231	func (w *worker) publish(res vmResult) {
		232	w.mu.Lock()
		233	defer w.mu.Unlock()
		234	w.result = res
		235	}
		236
221	func (w *worker) stop() {	237	func (w *worker) stop() {
222	w.mu.Lock()	238	w.mu.Lock()
223	w.stopped = true	239	w.stopped = true

internal/agent/reconcile/worker_test.go

Old		New
@@ -13,8 +13,8 @@ import (
13	// wedge returns an Images func that blocks every fetch until release is closed,	13	// wedge returns an Images func that blocks every fetch until release is closed,
14	// and signals arrived once per call so a test can observe how many VMs are	14	// and signals arrived once per call so a test can observe how many VMs are
15	// inside a slow operation at the same moment.	15	// inside a slow operation at the same moment.
16	func wedge(arrived chan<- struct{}, release <-chan struct{}) func(context.Context, string, string) (string, error) {	16	func wedge(arrived chan<- struct{}, release <-chan struct{}) func(context.Context, string, string, func(int64, int64)) (string, error) {
17	return func(ctx context.Context, url, sha string) (string, error) {	17	return func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
18	select {	18	select {
19	case arrived <- struct{}{}:	19	case arrived <- struct{}{}:
20	default:	20	default:
@@ -65,9 +65,16 @@ func TestBusyVMDoesNotBlockTheReport(t *testing.T) {
65	t.Fatal("worker never started the create")	65	t.Fatal("worker never started the create")
66	}	66	}
67		67
68	// The next tick must come back promptly even though vm1 is still wedged.	68	// The next tick must come back promptly even though vm1 is still wedged —
		69	// and it carries the wedged VM's row, saying what the wedge is. A create
		70	// publishes as it goes now (see worker.publish), which is precisely the case
		71	// this test wedges: before, a VM's slowest minutes were a report with no row
		72	// for it at all.
69	rep := stepNoWait(t, f, snap(1, vm("vm1")))	73	rep := stepNoWait(t, f, snap(1, vm("vm1")))
70	assert.Nil(t, findVM(rep, "vm1"), "a VM that has never published contributes no row")	74	row := findVM(rep, "vm1")
		75	require.NotNil(t, row, "a wedged create must still be in the report")
		76	assert.Equal(t, "creating", row.GetPhase())
		77	assert.Equal(t, "downloading image", row.GetStatusDetail(), "the row says which step is wedged")
71		78
72	// Once the operation completes the VM publishes and appears in the report.	79	// Once the operation completes the VM publishes and appears in the report.
73	close(release)	80	close(release)
@@ -170,7 +177,7 @@ func TestCreateConcurrencyIsCapped(t *testing.T) {
170		177
171	entered := make(chan struct{}, 3)	178	entered := make(chan struct{}, 3)
172	release := make(chan struct{})	179	release := make(chan struct{})
173	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {	180	f.eng.Images = func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
174	entered <- struct{}{}	181	entered <- struct{}{}
175	<-release	182	<-release
176	return "/cache/x.raw", nil	183	return "/cache/x.raw", nil

internal/agent/syncclient/client_test.go

Old		New
@@ -217,7 +217,7 @@ func newClient(t testing.T, addr, fp, hostID, cred string) Client {
217	require.NoError(t, agentSt.SaveIdentity(id))	217	require.NoError(t, agentSt.SaveIdentity(id))
218	engine := &reconcile.Engine{	218	engine := &reconcile.Engine{
219	St: agentSt, Prov: noopProv{},	219	St: agentSt, Prov: noopProv{},
220	Images: func(context.Context, string, string) (string, error) { return "/x.raw", nil },	220	Images: func(context.Context, string, string, func(int64, int64)) (string, error) { return "/x.raw", nil },
221	Seed: func(string, seed.Params) error { return nil },	221	Seed: func(string, seed.Params) error { return nil },
222	HostKey: state.LoadOrCreateHostKey,	222	HostKey: state.LoadOrCreateHostKey,
223	// CIDR/grace not exercised by these tests.	223	// CIDR/grace not exercised by these tests.

internal/mcpserver/tools.go

Old		New
@@ -266,7 +266,7 @@ func (t *Tools) VMCreate(ctx context.Context, in VMCreateIn) (VMCreateOut, error
266	// tool's "do not assume failure" contract. Only real cancellation (ctx.Done)	266	// tool's "do not assume failure" contract. Only real cancellation (ctx.Done)
267	// aborts immediately.	267	// aborts immediately.
268	func (t *Tools) waitReady(ctx context.Context, id, name string, deadline time.Time) (string, error) {	268	func (t *Tools) waitReady(ctx context.Context, id, name string, deadline time.Time) (string, error) {
269	last := ""	269	last, detail := "", ""
270	sawListing := false	270	sawListing := false
271	var lastErr error	271	var lastErr error
272	for time.Now().Before(deadline) {	272	for time.Now().Before(deadline) {
@@ -279,13 +279,13 @@ func (t *Tools) waitReady(ctx context.Context, id, name string, deadline time.Ti
279	if vm.ID != id {	279	if vm.ID != id {
280	continue	280	continue
281	}	281	}
282	last = vm.Lifecycle	282	last, detail = vm.Lifecycle, vm.StatusDetail
283	if vm.Lifecycle == "ready" && vm.AssignedIP != "" {	283	if vm.Lifecycle == "ready" && vm.AssignedIP != "" {
284	return vm.AssignedIP, nil	284	return vm.AssignedIP, nil
285	}	285	}
286	}	286	}
287	}	287	}
288	reportProgress(ctx, fmt.Sprintf("creating vm %s: %s", name, lifecycleOrUnknown(last)))	288	reportProgress(ctx, fmt.Sprintf("creating vm %s: %s", name, lifecycleOrUnknown(last, detail)))
289	select {	289	select {
290	case <-ctx.Done():	290	case <-ctx.Done():
291	return "", ctx.Err()	291	return "", ctx.Err()
@@ -300,10 +300,18 @@ func (t *Tools) waitReady(ctx context.Context, id, name string, deadline time.Ti
300		300
301	// lifecycleOrUnknown words a lifecycle for a progress message, covering the	301	// lifecycleOrUnknown words a lifecycle for a progress message, covering the
302	// first poll or two where the control plane has not listed the VM yet.	302	// first poll or two where the control plane has not listed the VM yet.
303	func lifecycleOrUnknown(lifecycle string) string {	303	//
		304	// The host's own account of what it is doing rides along when there is one.
		305	// "creating" is a single word for minutes of image, disk and boot, and a caller
		306	// watching this stream has no other way to tell a download in progress from a
		307	// create that is going nowhere.
		308	func lifecycleOrUnknown(lifecycle, detail string) string {
304	if lifecycle == "" {	309	if lifecycle == "" {
305	return "not listed yet"	310	return "not listed yet"
306	}	311	}
		312	if detail != "" {
		313	return lifecycle + ", " + detail
		314	}
307	return lifecycle	315	return lifecycle
308	}	316	}
309		317

internal/pb/sync.pb.go

Old		New
@@ -610,6 +610,15 @@ type ActualVM struct {
610	// every report for as long as the VM exists (level-triggered), so a lost	610	// every report for as long as the VM exists (level-triggered), so a lost
611	// snapshot or a control-plane restart re-certifies without operator action.	611	// snapshot or a control-plane restart re-certifies without operator action.
612	SshHostPubkey string `protobuf:"bytes,6,opt,name=ssh_host_pubkey,json=sshHostPubkey,proto3" json:"ssh_host_pubkey,omitempty"`	612	SshHostPubkey string `protobuf:"bytes,6,opt,name=ssh_host_pubkey,json=sshHostPubkey,proto3" json:"ssh_host_pubkey,omitempty"`
		613	// What this VM's host is doing about it right now, in the host's own words:
		614	// "downloading image 1.2/3.7 GiB", "preparing root disk", "booting". A create
		615	// is minutes of work behind one word, and the host is the only thing that can
		616	// see inside it — the control plane knows the VM is creating and nothing more.
		617	//
		618	// Free text for display, never parsed. Empty is the normal state of a settled
		619	// VM and means only "nothing to add": an agent that predates this field sends
		620	// it never, and a console reading it reads exactly what it read before.
		621	StatusDetail string `protobuf:"bytes,7,opt,name=status_detail,json=statusDetail,proto3" json:"status_detail,omitempty"`
613	unknownFields protoimpl.UnknownFields	622	unknownFields protoimpl.UnknownFields
614	sizeCache protoimpl.SizeCache	623	sizeCache protoimpl.SizeCache
615	}	624	}
@@ -686,6 +695,13 @@ func (x *ActualVM) GetSshHostPubkey() string {
686	return ""	695	return ""
687	}	696	}
688		697
		698	func (x *ActualVM) GetStatusDetail() string {
		699	if x != nil {
		700	return x.StatusDetail
		701	}
		702	return ""
		703	}
		704
689	type QuarantinedVM struct {	705	type QuarantinedVM struct {
690	state protoimpl.MessageState `protogen:"open.v1"`	706	state protoimpl.MessageState `protogen:"open.v1"`
691	VmId string `protobuf:"bytes,1,opt,name=vm_id,json=vmId,proto3" json:"vm_id,omitempty"`	707	VmId string `protobuf:"bytes,1,opt,name=vm_id,json=vmId,proto3" json:"vm_id,omitempty"`
@@ -1494,6 +1510,7 @@ type ExposureActual struct {
1494	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`	1510	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
1495	State string `protobuf:"bytes,2,opt,name=state,proto3" json:"state,omitempty"` // "active"\|"failed"	1511	State string `protobuf:"bytes,2,opt,name=state,proto3" json:"state,omitempty"` // "active"\|"failed"
1496	Reason string `protobuf:"bytes,3,opt,name=reason,proto3" json:"reason,omitempty"` // the OS error, when failed; a bound port's ongoing trouble otherwise	1512	Reason string `protobuf:"bytes,3,opt,name=reason,proto3" json:"reason,omitempty"` // the OS error, when failed; a bound port's ongoing trouble otherwise
		1513	Sessions *ExposureSessions `protobuf:"bytes,4,opt,name=sessions,proto3" json:"sessions,omitempty"`
1497	unknownFields protoimpl.UnknownFields	1514	unknownFields protoimpl.UnknownFields
1498	sizeCache protoimpl.SizeCache	1515	sizeCache protoimpl.SizeCache
1499	}	1516	}
@@ -1549,6 +1566,95 @@ func (x *ExposureActual) GetReason() string {
1549	return ""	1566	return ""
1550	}	1567	}
1551		1568
		1569	func (x ExposureActual) GetSessions() ExposureSessions {
		1570	if x != nil {
		1571	return x.Sessions
		1572	}
		1573	return nil
		1574	}
		1575
		1576	// ExposureSessions is what one published port has carried since the agent
		1577	// serving it started: what it holds this instant, and what it has turned away
		1578	// or could not carry along the way.
		1579	//
		1580	// The two totals only ever go up, deliberately. Loss on a published port is
		1581	// intermittent — a burst at the cap, a guest that went away for a second — and
		1582	// a gauge sampled once per report would show a clean zero for every one of
		1583	// those bursts, which reads as a port with nothing wrong. A running total
		1584	// cannot hide a burst it has already counted.
		1585	//
		1586	// A MESSAGE rather than three scalars beside the state, so that its absence is
		1587	// readable. An agent that predates these counters reports none, which is not
		1588	// the same fact as an exposure that has counted zero; the console says nothing
		1589	// rather than presenting a zero it was never told.
		1590	type ExposureSessions struct {
		1591	state protoimpl.MessageState `protogen:"open.v1"`
		1592	// active is what the port holds right now — spliced TCP connections, or live
		1593	// UDP client sessions. The one gauge here, and the only number that goes down.
		1594	Active int64 `protobuf:"varint,1,opt,name=active,proto3" json:"active,omitempty"`
		1595	// refused counts callers turned away because the port was already at its cap:
		1596	// a TCP connection accepted and closed, a UDP datagram dropped without
		1597	// starting a session.
		1598	Refused int64 `protobuf:"varint,2,opt,name=refused,proto3" json:"refused,omitempty"`
		1599	// dropped counts conversations this host could not carry to the guest — no
		1600	// guest address yet, a dial the guest refused, a datagram that would not send
		1601	// — as opposed to ones the cap refused.
		1602	Dropped int64 `protobuf:"varint,3,opt,name=dropped,proto3" json:"dropped,omitempty"`
		1603	unknownFields protoimpl.UnknownFields
		1604	sizeCache protoimpl.SizeCache
		1605	}
		1606
		1607	func (x *ExposureSessions) Reset() {
		1608	*x = ExposureSessions{}
		1609	mi := &file_proto_eitri_v1_sync_proto_msgTypes[18]
		1610	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
		1611	ms.StoreMessageInfo(mi)
		1612	}
		1613
		1614	func (x *ExposureSessions) String() string {
		1615	return protoimpl.X.MessageStringOf(x)
		1616	}
		1617
		1618	func (*ExposureSessions) ProtoMessage() {}
		1619
		1620	func (x *ExposureSessions) ProtoReflect() protoreflect.Message {
		1621	mi := &file_proto_eitri_v1_sync_proto_msgTypes[18]
		1622	if x != nil {
		1623	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
		1624	if ms.LoadMessageInfo() == nil {
		1625	ms.StoreMessageInfo(mi)
		1626	}
		1627	return ms
		1628	}
		1629	return mi.MessageOf(x)
		1630	}
		1631
		1632	// Deprecated: Use ExposureSessions.ProtoReflect.Descriptor instead.
		1633	func (*ExposureSessions) Descriptor() ([]byte, []int) {
		1634	return file_proto_eitri_v1_sync_proto_rawDescGZIP(), []int{18}
		1635	}
		1636
		1637	func (x *ExposureSessions) GetActive() int64 {
		1638	if x != nil {
		1639	return x.Active
		1640	}
		1641	return 0
		1642	}
		1643
		1644	func (x *ExposureSessions) GetRefused() int64 {
		1645	if x != nil {
		1646	return x.Refused
		1647	}
		1648	return 0
		1649	}
		1650
		1651	func (x *ExposureSessions) GetDropped() int64 {
		1652	if x != nil {
		1653	return x.Dropped
		1654	}
		1655	return 0
		1656	}
		1657
1552	var File_proto_eitri_v1_sync_proto protoreflect.FileDescriptor	1658	var File_proto_eitri_v1_sync_proto protoreflect.FileDescriptor
1553		1659
1554	const file_proto_eitri_v1_sync_proto_rawDesc = "" +	1660	const file_proto_eitri_v1_sync_proto_rawDesc = "" +
@@ -1602,7 +1708,7 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1602	"\fdisk_used_gb\x18\a \x01(\x03R\n" +	1708	"\fdisk_used_gb\x18\a \x01(\x03R\n" +
1603	"diskUsedGb\x12 \n" +	1709	"diskUsedGb\x12 \n" +
1604	"\fdisk_free_gb\x18\b \x01(\x03R\n" +	1710	"\fdisk_free_gb\x18\b \x01(\x03R\n" +
1605	"diskFreeGb\"\xa2\x01\n" +	1711	"diskFreeGb\"\xc7\x01\n" +
1606	"\bActualVM\x12\x13\n" +	1712	"\bActualVM\x12\x13\n" +
1607	"\x05vm_id\x18\x01 \x01(\tR\x04vmId\x12\x14\n" +	1713	"\x05vm_id\x18\x01 \x01(\tR\x04vmId\x12\x14\n" +
1608	"\x05power\x18\x02 \x01(\tR\x05power\x12\x14\n" +	1714	"\x05power\x18\x02 \x01(\tR\x05power\x12\x14\n" +
@@ -1610,7 +1716,8 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1610	"\x02ip\x18\x04 \x01(\tR\x02ip\x12\x1d\n" +	1716	"\x02ip\x18\x04 \x01(\tR\x02ip\x12\x1d\n" +
1611	"\n" +	1717	"\n" +
1612	"last_error\x18\x05 \x01(\tR\tlastError\x12&\n" +	1718	"last_error\x18\x05 \x01(\tR\tlastError\x12&\n" +
1613	"\x0fssh_host_pubkey\x18\x06 \x01(\tR\rsshHostPubkey\"\x81\x01\n" +	1719	"\x0fssh_host_pubkey\x18\x06 \x01(\tR\rsshHostPubkey\x12#\n" +
		1720	"\rstatus_detail\x18\a \x01(\tR\fstatusDetail\"\x81\x01\n" +
1614	"\rQuarantinedVM\x12\x13\n" +	1721	"\rQuarantinedVM\x12\x13\n" +
1615	"\x05vm_id\x18\x01 \x01(\tR\x04vmId\x12\x12\n" +	1722	"\x05vm_id\x18\x01 \x01(\tR\x04vmId\x12\x12\n" +
1616	"\x04name\x18\x02 \x01(\tR\x04name\x12\x1f\n" +	1723	"\x04name\x18\x02 \x01(\tR\x04name\x12\x1f\n" +
@@ -1679,11 +1786,16 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1679	"\n" +	1786	"\n" +
1680	"guest_port\x18\x03 \x01(\rR\tguestPort\x12\x1b\n" +	1787	"guest_port\x18\x03 \x01(\rR\tguestPort\x12\x1b\n" +
1681	"\thost_port\x18\x04 \x01(\rR\bhostPort\x12\x1a\n" +	1788	"\thost_port\x18\x04 \x01(\rR\bhostPort\x12\x1a\n" +
1682	"\bprotocol\x18\x05 \x01(\tR\bprotocol\"N\n" +	1789	"\bprotocol\x18\x05 \x01(\tR\bprotocol\"\x86\x01\n" +
1683	"\x0eExposureActual\x12\x0e\n" +	1790	"\x0eExposureActual\x12\x0e\n" +
1684	"\x02id\x18\x01 \x01(\tR\x02id\x12\x14\n" +	1791	"\x02id\x18\x01 \x01(\tR\x02id\x12\x14\n" +
1685	"\x05state\x18\x02 \x01(\tR\x05state\x12\x16\n" +	1792	"\x05state\x18\x02 \x01(\tR\x05state\x12\x16\n" +
1686	"\x06reason\x18\x03 \x01(\tR\x06reasonB&Z$github.com/a73x/eitri/internal/pb;pbb\x06proto3"	1793	"\x06reason\x18\x03 \x01(\tR\x06reason\x126\n" +
		1794	"\bsessions\x18\x04 \x01(\v2\x1a.eitri.v1.ExposureSessionsR\bsessions\"^\n" +
		1795	"\x10ExposureSessions\x12\x16\n" +
		1796	"\x06active\x18\x01 \x01(\x03R\x06active\x12\x18\n" +
		1797	"\arefused\x18\x02 \x01(\x03R\arefused\x12\x18\n" +
		1798	"\adropped\x18\x03 \x01(\x03R\adroppedB&Z$github.com/a73x/eitri/internal/pb;pbb\x06proto3"
1687		1799
1688	var (	1800	var (
1689	file_proto_eitri_v1_sync_proto_rawDescOnce sync.Once	1801	file_proto_eitri_v1_sync_proto_rawDescOnce sync.Once
@@ -1697,7 +1809,7 @@ func file_proto_eitri_v1_sync_proto_rawDescGZIP() []byte {
1697	return file_proto_eitri_v1_sync_proto_rawDescData	1809	return file_proto_eitri_v1_sync_proto_rawDescData
1698	}	1810	}
1699		1811
1700	var file_proto_eitri_v1_sync_proto_msgTypes = make([]protoimpl.MessageInfo, 18)	1812	var file_proto_eitri_v1_sync_proto_msgTypes = make([]protoimpl.MessageInfo, 19)
1701	var file_proto_eitri_v1_sync_proto_goTypes = []any{	1813	var file_proto_eitri_v1_sync_proto_goTypes = []any{
1702	(*AgentMessage)(nil), // 0: eitri.v1.AgentMessage	1814	(*AgentMessage)(nil), // 0: eitri.v1.AgentMessage
1703	(*ServerMessage)(nil), // 1: eitri.v1.ServerMessage	1815	(*ServerMessage)(nil), // 1: eitri.v1.ServerMessage
@@ -1717,6 +1829,7 @@ var file_proto_eitri_v1_sync_proto_goTypes = []any{
1717	(*TCPOpened)(nil), // 15: eitri.v1.TCPOpened	1829	(*TCPOpened)(nil), // 15: eitri.v1.TCPOpened
1718	(*ExposureDesired)(nil), // 16: eitri.v1.ExposureDesired	1830	(*ExposureDesired)(nil), // 16: eitri.v1.ExposureDesired
1719	(*ExposureActual)(nil), // 17: eitri.v1.ExposureActual	1831	(*ExposureActual)(nil), // 17: eitri.v1.ExposureActual
		1832	(*ExposureSessions)(nil), // 18: eitri.v1.ExposureSessions
1720	}	1833	}
1721	var file_proto_eitri_v1_sync_proto_depIdxs = []int32{	1834	var file_proto_eitri_v1_sync_proto_depIdxs = []int32{
1722	2, // 0: eitri.v1.AgentMessage.hello:type_name -> eitri.v1.Hello	1835	2, // 0: eitri.v1.AgentMessage.hello:type_name -> eitri.v1.Hello
@@ -1736,11 +1849,12 @@ var file_proto_eitri_v1_sync_proto_depIdxs = []int32{
1736	9, // 14: eitri.v1.DesiredStateSnapshot.vms:type_name -> eitri.v1.VMDesired	1849	9, // 14: eitri.v1.DesiredStateSnapshot.vms:type_name -> eitri.v1.VMDesired
1737	11, // 15: eitri.v1.DesiredStateSnapshot.agent_upgrade:type_name -> eitri.v1.AgentUpgrade	1850	11, // 15: eitri.v1.DesiredStateSnapshot.agent_upgrade:type_name -> eitri.v1.AgentUpgrade
1738	16, // 16: eitri.v1.DesiredStateSnapshot.exposures:type_name -> eitri.v1.ExposureDesired	1851	16, // 16: eitri.v1.DesiredStateSnapshot.exposures:type_name -> eitri.v1.ExposureDesired
1739	17, // [17:17] is the sub-list for method output_type	1852	18, // 17: eitri.v1.ExposureActual.sessions:type_name -> eitri.v1.ExposureSessions
1740	17, // [17:17] is the sub-list for method input_type	1853	18, // [18:18] is the sub-list for method output_type
1741	17, // [17:17] is the sub-list for extension type_name	1854	18, // [18:18] is the sub-list for method input_type
1742	17, // [17:17] is the sub-list for extension extendee	1855	18, // [18:18] is the sub-list for extension type_name
1743	0, // [0:17] is the sub-list for field type_name	1856	18, // [18:18] is the sub-list for extension extendee
		1857	0, // [0:18] is the sub-list for field type_name
1744	}	1858	}
1745		1859
1746	func init() { file_proto_eitri_v1_sync_proto_init() }	1860	func init() { file_proto_eitri_v1_sync_proto_init() }
@@ -1765,7 +1879,7 @@ func file_proto_eitri_v1_sync_proto_init() {
1765	GoPackagePath: reflect.TypeOf(x{}).PkgPath(),	1879	GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
1766	RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_eitri_v1_sync_proto_rawDesc), len(file_proto_eitri_v1_sync_proto_rawDesc)),	1880	RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_eitri_v1_sync_proto_rawDesc), len(file_proto_eitri_v1_sync_proto_rawDesc)),
1767	NumEnums: 0,	1881	NumEnums: 0,
1768	NumMessages: 18,	1882	NumMessages: 19,
1769	NumExtensions: 0,	1883	NumExtensions: 0,
1770	NumServices: 0,	1884	NumServices: 0,
1771	},	1885	},

internal/server/api/api.go

Old		New
@@ -567,27 +567,28 @@ func (a API) handleListHosts(w http.ResponseWriter, r http.Request) {
567		567
568	// toVMResponse merges a durable VM row with live agent-reported actual-state	568	// toVMResponse merges a durable VM row with live agent-reported actual-state
569	// into the wire shape (types.VM).	569	// into the wire shape (types.VM).
570	func toVMResponse(vm store.VM, actualPower, phase string, destroyAt int64) types.VM {	570	func toVMResponse(vm store.VM, actualPower, phase, statusDetail string, destroyAt int64) types.VM {
571	return types.VM{	571	return types.VM{
572	ID: vm.ID,	572	ID: vm.ID,
573	HostID: vm.HostID,	573	HostID: vm.HostID,
574	Name: vm.Name,	574	Name: vm.Name,
575	ImageURL: vm.ImageURL,	575	ImageURL: vm.ImageURL,
576	VCPUs: vm.VCPUs,	576	VCPUs: vm.VCPUs,
577	MemMB: vm.MemMB,	577	MemMB: vm.MemMB,
578	DiskGB: vm.DiskGB,	578	DiskGB: vm.DiskGB,
579	PowerState: vm.PowerState,	579	PowerState: vm.PowerState,
580	Status: vm.Status,	580	Status: vm.Status,
581	LastError: vm.LastError,	581	LastError: vm.LastError,
582	AssignedIP: vm.AssignedIP,	582	AssignedIP: vm.AssignedIP,
583	CreatedAt: vm.CreatedAt,	583	CreatedAt: vm.CreatedAt,
584	Deleted: vm.DeletedAt != nil,	584	Deleted: vm.DeletedAt != nil,
585	ActualPower: actualPower,	585	ActualPower: actualPower,
586	Phase: phase,	586	Phase: phase,
587	DestroyAt: destroyAt,	587	StatusDetail: statusDetail,
588	Lifecycle: deriveLifecycle(vm, actualPower, phase),	588	DestroyAt: destroyAt,
589	InjectedKey: injectedKey(vm),	589	Lifecycle: deriveLifecycle(vm, actualPower, phase),
590	TrustedCAs: trustedCAs(vm),	590	InjectedKey: injectedKey(vm),
		591	TrustedCAs: trustedCAs(vm),
591	}	592	}
592	}	593	}
593		594
@@ -639,13 +640,14 @@ func (a *API) snapshotVMs(p Principal) ([]types.VM, error) {
639	func (a *API) buildVMResponses(vms []store.VM, states map[string]regState) []types.VM {	640	func (a *API) buildVMResponses(vms []store.VM, states map[string]regState) []types.VM {
640	out := make([]types.VM, len(vms))	641	out := make([]types.VM, len(vms))
641	for i, vm := range vms {	642	for i, vm := range vms {
642	var actualPower, phase string	643	var actualPower, phase, statusDetail string
643	var destroyAt int64	644	var destroyAt int64
644	if rs := states[vm.HostID]; rs.ok {	645	if rs := states[vm.HostID]; rs.ok {
645	for _, av := range rs.st.Report.VMs {	646	for _, av := range rs.st.Report.VMs {
646	if av.VMID == vm.ID {	647	if av.VMID == vm.ID {
647	actualPower = av.Power	648	actualPower = av.Power
648	phase = av.Phase	649	phase = av.Phase
		650	statusDetail = av.StatusDetail
649	break	651	break
650	}	652	}
651	}	653	}
@@ -658,7 +660,7 @@ func (a *API) buildVMResponses(vms []store.VM, states map[string]regState) []typ
658	}	660	}
659	}	661	}
660	}	662	}
661	out[i] = toVMResponse(vm, actualPower, phase, destroyAt)	663	out[i] = toVMResponse(vm, actualPower, phase, statusDetail, destroyAt)
662	}	664	}
663	return out	665	return out
664	}	666	}

internal/server/api/exposures.go

Old		New
@@ -72,24 +72,35 @@ func exposureProtocol(req types.CreateExposureRequest) string {
72	// exposureState folds a host's live report into one exposure's state. An	72	// exposureState folds a host's live report into one exposure's state. An
73	// exposure the host has not reported on is "pending": the grant exists, and	73	// exposure the host has not reported on is "pending": the grant exists, and
74	// nothing has said what the host made of it yet.	74	// nothing has said what the host made of it yet.
75	func exposureState(statuses []registry.ExposureStatus, id string) (state, reason string) {	75	func exposureState(statuses []registry.ExposureStatus, id string) (state, reason string, sessions *types.ExposureSessions) {
76	for _, s := range statuses {	76	for _, s := range statuses {
77	if s.ID == id {	77	if s.ID == id {
78	return s.State, s.Reason	78	return s.State, s.Reason, toExposureSessions(s.Sessions)
79	}	79	}
80	}	80	}
81	return "pending", ""	81	return "pending", "", nil
		82	}
		83
		84	// toExposureSessions maps a host's reported counters to the wire shape, and
		85	// keeps nil meaning nil the whole way: an exposure nobody has reported on, and
		86	// one served by an agent that predates the counters, both say nothing rather
		87	// than saying zero.
		88	func toExposureSessions(s registry.ExposureSessions) types.ExposureSessions {
		89	if s == nil {
		90	return nil
		91	}
		92	return &types.ExposureSessions{Active: s.Active, Refused: s.Refused, Dropped: s.Dropped}
82	}	93	}
83		94
84	// toExposureResponse merges a durable exposure row with its host's address and	95	// toExposureResponse merges a durable exposure row with its host's address and
85	// live reported state into the wire shape.	96	// live reported state into the wire shape.
86	func toExposureResponse(e store.Exposure, hostAddr string, statuses []registry.ExposureStatus) types.Exposure {	97	func toExposureResponse(e store.Exposure, hostAddr string, statuses []registry.ExposureStatus) types.Exposure {
87	state, reason := exposureState(statuses, e.ID)	98	state, reason, sessions := exposureState(statuses, e.ID)
88	return types.Exposure{	99	return types.Exposure{
89	ID: e.ID, VMID: e.VMID, HostID: e.HostID,	100	ID: e.ID, VMID: e.VMID, HostID: e.HostID,
90	GuestPort: e.GuestPort, HostPort: e.HostPort, HostAddr: hostAddr,	101	GuestPort: e.GuestPort, HostPort: e.HostPort, HostAddr: hostAddr,
91	Protocol: e.Protocol, Scope: e.Scope,	102	Protocol: e.Protocol, Scope: e.Scope,
92	State: state, Reason: reason, CreatedAt: e.CreatedAt,	103	State: state, Reason: reason, Sessions: sessions, CreatedAt: e.CreatedAt,
93	}	104	}
94	}	105	}
95		106

internal/server/api/narration_test.go

@@ @@ -0,0 +1,134 @@ @@
+package api
+import (
+	"encoding/json"
+	"testing"
+	"github.com/a73x/eitri/internal/server/registry"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// vmByID pulls one VM out of the list response.
+func vmByID(t *testing.T, ts string, id string) map[string]any {
+	t.Helper()
+	resp := do(t, "GET", ts+"/api/v1/vms", testPAT, nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var list []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&list))
+	for _, vm := range list {
+		if vm["id"] == id {
+			return vm
+		}
+	}
+	t.Fatalf("vm %s not in the listing", id)
+	return nil
+}
+// TestWhatAHostIsDoingReachesTheWire is the outbound leg of the narration: the
+// sentence the host reported is served beside the phase, so the console can put
+// it under a VM that would otherwise say only "creating".
+func TestWhatAHostIsDoingReachesTheWire(t *testing.T) {
+	ts, _, _, reg, _ := newServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	reg.UpdateReport(host["host_id"], registry.Report{
+		VMs: []registry.ActualVM{{
+			VMID: vmID, Power: "stopped", Phase: "creating",
+			StatusDetail: "downloading image 1.2/3.7 GiB",
+		}},
+	})
+	vm := vmByID(t, ts.URL, vmID)
+	assert.Equal(t, "creating", vm["phase"])
+	assert.Equal(t, "downloading image 1.2/3.7 GiB", vm["status_detail"])
+}
+// TestAVMOnASilentAgentReadsAsItAlwaysDid is the fielded-agent case end to end:
+// a host that reports no detail — an agent older than the field — serves an
+// empty one, and the console renders exactly what it rendered before.
+func TestAVMOnASilentAgentReadsAsItAlwaysDid(t *testing.T) {
+	ts, _, _, reg, _ := newServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	reg.UpdateReport(host["host_id"], registry.Report{
+		VMs: []registry.ActualVM{{VMID: vmID, Power: "stopped", Phase: "creating"}},
+	})
+	vm := vmByID(t, ts.URL, vmID)
+	assert.Equal(t, "creating", vm["phase"])
+	assert.Equal(t, "", vm["status_detail"])
+}
+// TestExposureSessionCountersReachTheWire pins the counters' outbound leg: what
+// the host counted is what the exposure row serves.
+func TestExposureSessionCountersReachTheWire(t *testing.T) {
+	ts, _, _, reg, _ := newServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080, "host_port": 30080})
+	require.Equal(t, 201, resp.StatusCode)
+	var created map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&created))
+	reg.UpdateReport(host["host_id"], registry.Report{
+		Exposures: []registry.ExposureStatus{{
+			ID: created["id"].(string), State: "active",
+			Sessions: &registry.ExposureSessions{Active: 7, Refused: 12, Dropped: 3},
+		}},
+	})
+	resp = do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT, nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var list []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&list))
+	require.Len(t, list, 1)
+	sessions, ok := list[0]["sessions"].(map[string]any)
+	require.True(t, ok, "a counted exposure carries its counters")
+	assert.Equal(t, float64(7), sessions["active"])
+	assert.Equal(t, float64(12), sessions["refused"])
+	assert.Equal(t, float64(3), sessions["dropped"])
+}
+// TestAnUncountedExposureServesNoCounters is the case that decides the console's
+// display rule. Two exposures nobody counted: one on an agent too old to count
+// (reported, no counters), one nothing has reported on at all. Both serve null
+// rather than zeros, because a row of zeros is a claim — "this port has turned
+// nobody away" — that nothing on the wire supports.
+func TestAnUncountedExposureServesNoCounters(t *testing.T) {
+	ts, _, _, reg, _ := newServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	newExposure := func(guestPort, hostPort int64) string {
+		t.Helper()
+		resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+			map[string]any{"guest_port": guestPort, "host_port": hostPort})
+		require.Equal(t, 201, resp.StatusCode)
+		var e map[string]any
+		require.NoError(t, json.NewDecoder(resp.Body).Decode(&e))
+		return e["id"].(string)
+	}
+	reported := newExposure(8080, 30080)
+	newExposure(9090, 30081) // never reported on
+	reg.UpdateReport(host["host_id"], registry.Report{
+		Exposures: []registry.ExposureStatus{{ID: reported, State: "active"}},
+	})
+	resp := do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT, nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var list []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&list))
+	require.Len(t, list, 2)
+	assert.Equal(t, "active", list[0]["state"])
+	assert.Nil(t, list[0]["sessions"], "an agent that counts nothing reports nothing")
+	assert.Equal(t, "pending", list[1]["state"])
+	assert.Nil(t, list[1]["sessions"], "an unreported exposure has no counters either")
+}

internal/server/api/testdata/exposure.golden.json

Old		New
@@ -10,6 +10,11 @@
10	"scope": "lan",	10	"scope": "lan",
11	"state": "failed",	11	"state": "failed",
12	"reason": "listen tcp 0.0.0.0:30080: bind: address already in use",	12	"reason": "listen tcp 0.0.0.0:30080: bind: address already in use",
13	"created_at": "2026-07-27T12:04:00Z"	13	"created_at": "2026-07-27T12:04:00Z",
		14	"sessions": {
		15	"active": 7,
		16	"refused": 12,
		17	"dropped": 3
		18	}
14	}	19	}
15	]	20	]

internal/server/api/testdata/snapshot.golden.json

Old		New
@@ -66,6 +66,7 @@
66	"deleted": true,	66	"deleted": true,
67	"actual_power": "stopped",	67	"actual_power": "stopped",
68	"phase": "creating",	68	"phase": "creating",
		69	"status_detail": "downloading image 1.2/3.7 GiB",
69	"destroy_at": 1785153600,	70	"destroy_at": 1785153600,
70	"lifecycle": "deleting",	71	"lifecycle": "deleting",
71	"injected_key": {	72	"injected_key": {

internal/server/api/testdata/vm.golden.json

Old		New
@@ -14,6 +14,7 @@
14	"deleted": true,	14	"deleted": true,
15	"actual_power": "stopped",	15	"actual_power": "stopped",
16	"phase": "creating",	16	"phase": "creating",
		17	"status_detail": "downloading image 1.2/3.7 GiB",
17	"destroy_at": 1785153600,	18	"destroy_at": 1785153600,
18	"lifecycle": "deleting",	19	"lifecycle": "deleting",
19	"injected_key": {	20	"injected_key": {

internal/server/api/types/types.go

Old		New
@@ -119,6 +119,16 @@ type VM struct {
119	Deleted bool `json:"deleted"`	119	Deleted bool `json:"deleted"`
120	ActualPower string `json:"actual_power"`	120	ActualPower string `json:"actual_power"`
121	Phase string `json:"phase"`	121	Phase string `json:"phase"`
		122	// StatusDetail is what this VM's host is doing about it right now, in the
		123	// host's own words: "downloading image 1.2/3.7 GiB", "preparing root disk",
		124	// "booting". It exists because `creating` is one word for minutes of work
		125	// that only the host can see inside.
		126	//
		127	// Free text for reading, never for branching: it is the host's sentence, and
		128	// the control plane neither parses nor validates it. Empty means "nothing to
		129	// add" — a settled VM, or a host too old to say — and a client shows nothing
		130	// rather than a placeholder.
		131	StatusDetail string `json:"status_detail"`
122	// DestroyAt is the unix-seconds deadline at which the agent will hard-destroy	132	// DestroyAt is the unix-seconds deadline at which the agent will hard-destroy
123	// this VM. It is only set while the VM is quarantined for teardown (deleted +	133	// this VM. It is only set while the VM is quarantined for teardown (deleted +
124	// guest stopped, awaiting the tombstone grace window); 0 in the normal case.	134	// guest stopped, awaiting the tombstone grace window); 0 in the normal case.
@@ -303,6 +313,27 @@ type Exposure struct {
303	State string `json:"state"`	313	State string `json:"state"`
304	Reason string `json:"reason"`	314	Reason string `json:"reason"`
305	CreatedAt time.Time `json:"created_at"`	315	CreatedAt time.Time `json:"created_at"`
		316	// Sessions is what the host says this port has carried, or null when the
		317	// host has not said — an exposure not yet reported on, or an agent older
		318	// than the counters. Null is not zero: a client shows nothing at all rather
		319	// than presenting zeros nobody reported.
		320	Sessions *ExposureSessions `json:"sessions"`
		321	}
		322
		323	// ExposureSessions is one published port's traffic as its host counts it.
		324	// Active is a gauge — conversations open this instant. Refused and Dropped are
		325	// running totals SINCE THAT AGENT STARTED, because what they count is momentary
		326	// and a gauge would read zero between two bursts of it. An agent restart resets
		327	// both, which is why a client says what they are counted from.
		328	type ExposureSessions struct {
		329	// Active is what the port holds right now: open TCP connections, or live UDP
		330	// client sessions.
		331	Active int64 `json:"active"`
		332	// Refused counts callers the port turned away because it was at its cap.
		333	Refused int64 `json:"refused"`
		334	// Dropped counts conversations the host could not carry to the guest: no
		335	// guest address yet, a refused dial, a datagram that would not send.
		336	Dropped int64 `json:"dropped"`
306	}	337	}
307		338
308	// The response shapes below replace handlers' inline map[string]string	339	// The response shapes below replace handlers' inline map[string]string

internal/server/api/wire_golden_test.go

Old		New
@@ -91,23 +91,24 @@ func TestWireGolden(t *testing.T) {
91	goldenCheck(t, "host", host)	91	goldenCheck(t, "host", host)
92		92
93	vm := types.VM{	93	vm := types.VM{
94	ID: "v-5678",	94	ID: "v-5678",
95	HostID: "h-1234",	95	HostID: "h-1234",
96	Name: "sandbox-abc123",	96	Name: "sandbox-abc123",
97	ImageURL: "https://images.example.com/resolute.img",	97	ImageURL: "https://images.example.com/resolute.img",
98	VCPUs: 2,	98	VCPUs: 2,
99	MemMB: 2048,	99	MemMB: 2048,
100	DiskGB: 10,	100	DiskGB: 10,
101	PowerState: "running",	101	PowerState: "running",
102	Status: "ready",	102	Status: "ready",
103	LastError: "boot timeout",	103	LastError: "boot timeout",
104	AssignedIP: "10.77.1.2",	104	AssignedIP: "10.77.1.2",
105	CreatedAt: base.Add(time.Minute),	105	CreatedAt: base.Add(time.Minute),
106	Deleted: true,	106	Deleted: true,
107	ActualPower: "stopped",	107	ActualPower: "stopped",
108	Phase: "creating",	108	Phase: "creating",
109	DestroyAt: 1785153600,	109	StatusDetail: "downloading image 1.2/3.7 GiB",
110	Lifecycle: "deleting",	110	DestroyAt: 1785153600,
		111	Lifecycle: "deleting",
111	InjectedKey: &types.InjectedKey{	112	InjectedKey: &types.InjectedKey{
112	Type: "ssh-ed25519",	113	Type: "ssh-ed25519",
113	Fingerprint: "SHA256:0000000000000000000000000000000000000000000",	114	Fingerprint: "SHA256:0000000000000000000000000000000000000000000",
@@ -278,6 +279,7 @@ func TestWireGolden(t *testing.T) {
278	Scope: "lan",	279	Scope: "lan",
279	State: "failed",	280	State: "failed",
280	Reason: "listen tcp 0.0.0.0:30080: bind: address already in use",	281	Reason: "listen tcp 0.0.0.0:30080: bind: address already in use",
		282	Sessions: &types.ExposureSessions{Active: 7, Refused: 12, Dropped: 3},
281	CreatedAt: base.Add(4 * time.Minute),	283	CreatedAt: base.Add(4 * time.Minute),
282	}})	284	}})
283	}	285	}

internal/server/registry/registry.go

Old		New
@@ -38,6 +38,10 @@ type Metrics struct {
38		38
39	type ActualVM struct {	39	type ActualVM struct {
40	VMID, Power, Phase, IP, LastError string	40	VMID, Power, Phase, IP, LastError string
		41	// StatusDetail is what the host is doing about this VM right now, in the
		42	// host's own words. Empty is the normal state of a settled VM, and is also
		43	// what an agent too old to say anything leaves behind.
		44	StatusDetail string
41	}	45	}
42		46
43	type QuarantinedVM struct {	47	type QuarantinedVM struct {
@@ -51,7 +55,19 @@ type QuarantinedVM struct {
51	// otherwise. Never persisted — like Metrics, it lives only while the host is	55	// otherwise. Never persisted — like Metrics, it lives only while the host is
52	// connected, and an exposure with no row here has simply not been reported on	56	// connected, and an exposure with no row here has simply not been reported on
53	// yet.	57	// yet.
54	type ExposureStatus struct{ ID, State, Reason string }	58	type ExposureStatus struct {
		59	ID, State, Reason string
		60	// Sessions is what the port has carried, or nil when the agent serving it
		61	// does not count — an agent older than the counters. Nil is deliberately
		62	// distinguishable from a zeroed struct: one is "nobody said", the other is
		63	// "nothing has happened", and only the second is a fact worth showing.
		64	Sessions *ExposureSessions
		65	}
		66
		67	// ExposureSessions is one published port's traffic as its host counts it:
		68	// Active right now, Refused and Dropped cumulatively since that agent started.
		69	// See ExposureSessions in the proto for why the two totals are not gauges.
		70	type ExposureSessions struct{ Active, Refused, Dropped int64 }
55		71
56	type Report struct {	72	type Report struct {
57	VMs []ActualVM	73	VMs []ActualVM

internal/server/syncsvc/narration_test.go

@@ @@ -0,0 +1,85 @@ @@
+package syncsvc
+import (
+	"testing"
+	"github.com/a73x/eitri/internal/pb"
+	"github.com/a73x/eitri/internal/server/registry"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// TestApplyReportCarriesWhatAHostIsDoing pins the inbound leg of the narration:
+// what a host says it is doing about a VM lands in the registry beside the
+// phase, where the VM API reads it.
+func TestApplyReportCarriesWhatAHostIsDoing(t *testing.T) {
+	f := setup(t)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{
+		Vms: []*pb.ActualVM{{
+			VmId: "vm1", Power: "stopped", Phase: "creating",
+			StatusDetail: "downloading image 1.2/3.7 GiB",
+		}},
+	})
+	got, ok := f.reg.Get(f.host.ID)
+	require.True(t, ok)
+	require.Len(t, got.Report.VMs, 1)
+	assert.Equal(t, "downloading image 1.2/3.7 GiB", got.Report.VMs[0].StatusDetail)
+}
+// TestApplyReportFromAnAgentThatSaysNothing is the fielded-agent case: an agent
+// that predates status_detail sends none, and the field is absent rather than
+// wrong. Everything about the VM must read exactly as it did before the field
+// existed.
+func TestApplyReportFromAnAgentThatSaysNothing(t *testing.T) {
+	f := setup(t)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{
+		Vms: []*pb.ActualVM{{VmId: "vm1", Power: "stopped", Phase: "creating"}},
+	})
+	got, ok := f.reg.Get(f.host.ID)
+	require.True(t, ok)
+	require.Len(t, got.Report.VMs, 1)
+	assert.Equal(t, "creating", got.Report.VMs[0].Phase)
+	assert.Empty(t, got.Report.VMs[0].StatusDetail, "an agent that says nothing must not be quoted")
+}
+// TestApplyReportCountsExposureSessions pins the counters' inbound leg, gauge
+// and totals together.
+func TestApplyReportCountsExposureSessions(t *testing.T) {
+	f := setup(t)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{
+		Exposures: []*pb.ExposureActual{{
+			Id: "e1", State: "active",
+			Sessions: &pb.ExposureSessions{Active: 7, Refused: 12, Dropped: 3},
+		}},
+	})
+	got, ok := f.reg.Get(f.host.ID)
+	require.True(t, ok)
+	require.Len(t, got.Report.Exposures, 1)
+	require.NotNil(t, got.Report.Exposures[0].Sessions)
+	assert.Equal(t, registry.ExposureSessions{Active: 7, Refused: 12, Dropped: 3},
+		*got.Report.Exposures[0].Sessions)
+}
+// TestApplyReportFromAnAgentThatCountsNothing is the other half of the fielded-
+// agent case, and the one where zero would be a lie: an agent that reports no
+// counters leaves nil, NOT a zeroed struct. "Nobody said" and "nothing has
+// happened" are different answers, and only the second one is a fact.
+func TestApplyReportFromAnAgentThatCountsNothing(t *testing.T) {
+	f := setup(t)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{
+		Exposures: []*pb.ExposureActual{{Id: "e1", State: "active"}},
+	})
+	got, ok := f.reg.Get(f.host.ID)
+	require.True(t, ok)
+	require.Len(t, got.Report.Exposures, 1)
+	assert.Equal(t, "active", got.Report.Exposures[0].State)
+	assert.Nil(t, got.Report.Exposures[0].Sessions, "an uncounted port must not report zeros")
+}

internal/server/syncsvc/syncsvc.go

Old		New
@@ -617,11 +617,12 @@ func toRegistryVMs(in []*pb.ActualVM) []registry.ActualVM {
617	out := make([]registry.ActualVM, 0, len(in))	617	out := make([]registry.ActualVM, 0, len(in))
618	for _, v := range in {	618	for _, v := range in {
619	out = append(out, registry.ActualVM{	619	out = append(out, registry.ActualVM{
620	VMID: v.GetVmId(),	620	VMID: v.GetVmId(),
621	Power: v.GetPower(),	621	Power: v.GetPower(),
622	Phase: v.GetPhase(),	622	Phase: v.GetPhase(),
623	IP: v.GetIp(),	623	IP: v.GetIp(),
624	LastError: v.GetLastError(),	624	LastError: v.GetLastError(),
		625	StatusDetail: v.GetStatusDetail(),
625	})	626	})
626	}	627	}
627	return out	628	return out
@@ -655,11 +656,25 @@ func toRegistryExposures(in []*pb.ExposureActual) []registry.ExposureStatus {
655	for _, e := range in {	656	for _, e := range in {
656	out = append(out, registry.ExposureStatus{	657	out = append(out, registry.ExposureStatus{
657	ID: e.GetId(), State: e.GetState(), Reason: e.GetReason(),	658	ID: e.GetId(), State: e.GetState(), Reason: e.GetReason(),
		659	Sessions: toRegistrySessions(e.GetSessions()),
658	})	660	})
659	}	661	}
660	return out	662	return out
661	}	663	}
662		664
		665	// toRegistrySessions maps an exposure's reported counters, keeping the one
		666	// distinction the whole field exists for: an agent that reports no counters
		667	// (nil) is not an exposure that has counted zero, and it must not become one on
		668	// the way through.
		669	func toRegistrySessions(s pb.ExposureSessions) registry.ExposureSessions {
		670	if s == nil {
		671	return nil
		672	}
		673	return &registry.ExposureSessions{
		674	Active: s.GetActive(), Refused: s.GetRefused(), Dropped: s.GetDropped(),
		675	}
		676	}
		677
663	// toRegistryCapacity maps reported capacity (nil → zero value).	678	// toRegistryCapacity maps reported capacity (nil → zero value).
664	func toRegistryCapacity(c *pb.Capacity) registry.Capacity {	679	func toRegistryCapacity(c *pb.Capacity) registry.Capacity {
665	if c == nil {	680	if c == nil {

internal/transport/contract_test.go

Old		New
@@ -7,6 +7,7 @@ import (
7	"github.com/a73x/eitri/internal/pb"	7	"github.com/a73x/eitri/internal/pb"
8	"github.com/stretchr/testify/assert"	8	"github.com/stretchr/testify/assert"
9	"github.com/stretchr/testify/require"	9	"github.com/stretchr/testify/require"
		10	"google.golang.org/protobuf/encoding/protowire"
10	"google.golang.org/protobuf/proto"	11	"google.golang.org/protobuf/proto"
11	)	12	)
12		13
@@ -21,7 +22,12 @@ func TestAgentMessageReportRoundTrip(t *testing.T) {
21	Vms: []*pb.ActualVM{	22	Vms: []*pb.ActualVM{
22	{VmId: "vm-1", Power: "on", Phase: "running", Ip: "10.0.0.5", LastError: ""},	23	{VmId: "vm-1", Power: "on", Phase: "running", Ip: "10.0.0.5", LastError: ""},
23	{VmId: "vm-2", Power: "off", Phase: "stopped"},	24	{VmId: "vm-2", Power: "off", Phase: "stopped"},
		25	{VmId: "vm-3", Power: "off", Phase: "creating", StatusDetail: "downloading image 1.2/3.7 GiB"},
24	},	26	},
		27	Exposures: []*pb.ExposureActual{{
		28	Id: "e-1", State: "active",
		29	Sessions: &pb.ExposureSessions{Active: 7, Refused: 12, Dropped: 3},
		30	}},
25	Destroyed: []string{"vm-old"},	31	Destroyed: []string{"vm-old"},
26	Quarantined: []*pb.QuarantinedVM{{VmId: "vm-q", Name: "q", VmspecJson: []byte(`{"k":1}`), DestroyAtUnix: 1750000000}},	32	Quarantined: []*pb.QuarantinedVM{{VmId: "vm-q", Name: "q", VmspecJson: []byte(`{"k":1}`), DestroyAtUnix: 1750000000}},
27	Capacity: &pb.Capacity{Vcpus: 16, MemMb: 32768, DiskGb: 500},	33	Capacity: &pb.Capacity{Vcpus: 16, MemMb: 32768, DiskGb: 500},
@@ -50,6 +56,46 @@ func TestServerMessageSnapshotRoundTrip(t *testing.T) {
50	assert.True(t, proto.Equal(in, out), "snapshot did not survive round-trip:\n in=%v\nout=%v", in, out)	56	assert.True(t, proto.Equal(in, out), "snapshot did not survive round-trip:\n in=%v\nout=%v", in, out)
51	}	57	}
52		58
		59	// TestAReportFromANewerAgentDecodesOnAnOlderServer pins the rule the field
		60	// numbering exists to serve. A mixed fleet is the normal state during a
		61	// rollout, so a peer that has never heard of a field must skip past it and read
		62	// everything else — not reject the frame, and not misread the field it does
		63	// know. Field 99 is nothing in this schema and stands in for whatever a later
		64	// release adds beside status_detail.
		65	func TestAReportFromANewerAgentDecodesOnAnOlderServer(t *testing.T) {
		66	raw, err := proto.Marshal(&pb.ActualVM{
		67	VmId: "vm-1", Power: "off", Phase: "creating",
		68	StatusDetail: "downloading image 1.2/3.7 GiB",
		69	})
		70	require.NoError(t, err)
		71	raw = protowire.AppendTag(raw, 99, protowire.BytesType)
		72	raw = protowire.AppendString(raw, "a field from a release this peer predates")
		73
		74	var got pb.ActualVM
		75	require.NoError(t, proto.Unmarshal(raw, &got), "an unknown field must not fail the frame")
		76	assert.Equal(t, "vm-1", got.GetVmId())
		77	assert.Equal(t, "creating", got.GetPhase())
		78	assert.Equal(t, "downloading image 1.2/3.7 GiB", got.GetStatusDetail())
		79	}
		80
		81	// TestAReportFromAnOlderAgentReadsAsSilence is the same rule from the other
		82	// side, and the one the console's display rules rest on: an agent that predates
		83	// these fields sends neither, and neither may arrive as a value. An absent
		84	// status_detail is empty (nothing to add) and absent counters are NIL — not a
		85	// zeroed struct claiming this port has turned nobody away.
		86	func TestAReportFromAnOlderAgentReadsAsSilence(t *testing.T) {
		87	raw, err := proto.Marshal(&pb.ActualStateReport{
		88	Vms: []*pb.ActualVM{{VmId: "vm-1", Power: "off", Phase: "creating"}},
		89	Exposures: []*pb.ExposureActual{{Id: "e-1", State: "active"}},
		90	})
		91	require.NoError(t, err)
		92
		93	var got pb.ActualStateReport
		94	require.NoError(t, proto.Unmarshal(raw, &got))
		95	assert.Empty(t, got.GetVms()[0].GetStatusDetail())
		96	assert.Nil(t, got.GetExposures()[0].GetSessions(), "an uncounted port must not decode as zeros")
		97	}
		98
53	// roundTrip frames in and reads it back into out.	99	// roundTrip frames in and reads it back into out.
54	func roundTrip(t *testing.T, in, out proto.Message) {	100	func roundTrip(t *testing.T, in, out proto.Message) {
55	t.Helper()	101	t.Helper()

proto/eitri/v1/sync.proto

@@ @@ -76,6 +76,15 @@ message ActualVM { @@
+  // every report for as long as the VM exists (level-triggered), so a lost
+  // snapshot or a control-plane restart re-certifies without operator action.
+  string ssh_host_pubkey = 6;
+  // What this VM's host is doing about it right now, in the host's own words:
+  // "downloading image 1.2/3.7 GiB", "preparing root disk", "booting". A create
+  // is minutes of work behind one word, and the host is the only thing that can
+  // see inside it — the control plane knows the VM is creating and nothing more.
+  //
+  // Free text for display, never parsed. Empty is the normal state of a settled
+  // VM and means only "nothing to add": an agent that predates this field sends
+  // it never, and a console reading it reads exactly what it read before.
+  string status_detail = 7;
+message QuarantinedVM {
@@ @@ -227,4 +236,33 @@ message ExposureActual { @@
+  string id = 1;
+  string state = 2;   // "active"|"failed"
+  string reason = 3;  // the OS error, when failed; a bound port's ongoing trouble otherwise
+  ExposureSessions sessions = 4;
+}
+// ExposureSessions is what one published port has carried since the agent
+// serving it started: what it holds this instant, and what it has turned away
+// or could not carry along the way.
+//
+// The two totals only ever go up, deliberately. Loss on a published port is
+// intermittent — a burst at the cap, a guest that went away for a second — and
+// a gauge sampled once per report would show a clean zero for every one of
+// those bursts, which reads as a port with nothing wrong. A running total
+// cannot hide a burst it has already counted.
+//
+// A MESSAGE rather than three scalars beside the state, so that its absence is
+// readable. An agent that predates these counters reports none, which is not
+// the same fact as an exposure that has counted zero; the console says nothing
+// rather than presenting a zero it was never told.
+message ExposureSessions {
+  // active is what the port holds right now — spliced TCP connections, or live
+  // UDP client sessions. The one gauge here, and the only number that goes down.
+  int64 active = 1;
+  // refused counts callers turned away because the port was already at its cap:
+  // a TCP connection accepted and closed, a UDP datagram dropped without
+  // starting a session.
+  int64 refused = 2;
+  // dropped counts conversations this host could not carry to the guest — no
+  // guest address yet, a dial the guest refused, a datagram that would not send
+  // — as opposed to ones the cap refused.
+  int64 dropped = 3;

web/src/lib/api-types.ts

Old		New
@@ -1579,9 +1579,15 @@ export interface components {
1579	protocol: string;	1579	protocol: string;
1580	reason: string;	1580	reason: string;
1581	scope: string;	1581	scope: string;
		1582	sessions?: components["schemas"]["ExposureSessions"] \| null;
1582	state: string;	1583	state: string;
1583	vm_id: string;	1584	vm_id: string;
1584	};	1585	};
		1586	ExposureSessions: {
		1587	active: number;
		1588	dropped: number;
		1589	refused: number;
		1590	};
1585	Host: {	1591	Host: {
1586	agent_update_available: boolean;	1592	agent_update_available: boolean;
1587	agent_version: string;	1593	agent_version: string;
@@ -1697,6 +1703,7 @@ export interface components {
1697	phase: string;	1703	phase: string;
1698	power_state: string;	1704	power_state: string;
1699	status: string;	1705	status: string;
		1706	status_detail: string;
1700	trusted_cas?: components["schemas"]["TrustedCA"][] \| null;	1707	trusted_cas?: components["schemas"]["TrustedCA"][] \| null;
1701	vcpus: number;	1708	vcpus: number;
1702	};	1709	};

web/src/lib/fleet.svelte.ts

@@ @@ -490,6 +490,22 @@ export function vmPower(vm: VM): string { @@
+	return vm.actual_power || vm.power_state;
+/** vmDetail is the sentence to show beneath a VM's status: what its host is
+ *  doing about it right now, in the host's own words.
+ *
+ *  Only a VM that is still being created gets one. The sentence describes work
+ *  in flight — "downloading image 1.2/3.7 GiB" — and a settled VM showing the
+ *  last thing its host was doing would be presenting the past as the present.
+ *  Everything else returns '' and the row is not drawn at all.
+ *
+ *  An empty detail is the normal case in two situations that look identical
+ *  here and should: a host with nothing to add, and a host running an agent old
+ *  enough that it never sends the field. Neither gets a placeholder. */
+export function vmDetail(vm: VM): string {
+	if (vmStatus(vm) !== 'creating') return '';
+	return vm.status_detail ?? '';
+}
+/** vmIP is the assigned IP, or an em-dash placeholder when unassigned. */
+export function vmIP(vm: VM): string {
+	return vm.assigned_ip || '—';
+	return addr.includes(':') ? `[${addr}]:${port}` : `${addr}:${port}`;
+/** sessionSummary is what an exposure row says about the traffic its port has
+ *  carried, or '' when the row should say nothing at all.
+ *
+ *  Saying nothing is the interesting half. The counters are absent for an
+ *  exposure no host has reported on yet, and for one served by an agent that
+ *  predates them — and in both cases the honest answer is silence. Rendering
+ *  zeros instead would state that this port has turned nobody away, which is a
+ *  claim nothing on the wire supports; a real zero, reported as zero, says
+ *  exactly that and is worth showing.
+ *
+ *  "since agent start" is not decoration. The two totals reset when the agent
+ *  restarts, so a number read without that clause is a number over an unknown
+ *  window. */
+export function sessionSummary(e: Exposure): string {
+	const s = e.sessions;
+	if (!s) return '';
+	return `${s.active} open · ${s.refused} refused, ${s.dropped} dropped since agent start`;
+}
+/** CapacityReading is one allocation read against the capacity its host
+ *  reports: what to draw, and what to say about the difference. */
+export type CapacityReading = {

web/src/lib/fleet.test.ts

Old		New
@@ -4,11 +4,14 @@ import {
4	fleet,	4	fleet,
5	hostBundle,	5	hostBundle,
6	joinCommands,	6	joinCommands,
		7	sessionSummary,
7	upgradeAge,	8	upgradeAge,
8	upgradeStuck,	9	upgradeStuck,
9	upgradeStuckNote,	10	upgradeStuckNote,
		11	vmDetail,
10	vmTrustStale,	12	vmTrustStale,
11	UPGRADE_STUCK_S,	13	UPGRADE_STUCK_S,
		14	type Exposure,
12	type Host,	15	type Host,
13	type UserCA,	16	type UserCA,
14	type VM	17	type VM
@@ -44,11 +47,38 @@ function vmTrusting(trusted_cas: TrustedCA[] \| null): VM {
44	phase: 'ready',	47	phase: 'ready',
45	power_state: 'running',	48	power_state: 'running',
46	status: 'ready',	49	status: 'ready',
		50	status_detail: '',
47	trusted_cas,	51	trusted_cas,
48	vcpus: 1	52	vcpus: 1
49	};	53	};
50	}	54	}
51		55
		56	/** vmCreating is a VM mid-create, carrying whatever its host last said it was
		57	* doing. Pass '' for a host with nothing to add — or one whose agent predates
		58	* the field and never sends it. */
		59	function vmCreating(status_detail: string): VM {
		60	return { ...vmTrusting(null), lifecycle: 'creating', phase: 'creating', status_detail };
		61	}
		62
		63	/** exposure is one published port as its row sees it. sessions is null for a
		64	* port nobody has counted: unreported, or served by an older agent. */
		65	function exposure(sessions: Exposure['sessions']): Exposure {
		66	return {
		67	created_at: '2026-08-11T09:00:00Z',
		68	guest_port: 8080,
		69	host_addr: '192.168.0.190',
		70	host_id: 'host-1',
		71	host_port: 30080,
		72	id: 'x-1',
		73	protocol: 'tcp',
		74	reason: '',
		75	scope: 'lan',
		76	sessions,
		77	state: 'active',
		78	vm_id: 'vm-1'
		79	};
		80	}
		81
52	beforeEach(() => {	82	beforeEach(() => {
53	fleet.userCAs = [];	83	fleet.userCAs = [];
54	fleet.userCAsLoaded = false;	84	fleet.userCAsLoaded = false;
@@ -334,3 +364,43 @@ describe('joinCommands', () => {
334	expect(cmds).not.toContain('${V}');	364	expect(cmds).not.toContain('${V}');
335	});	365	});
336	});	366	});
		367
		368	describe('vmDetail', () => {
		369	test('a creating VM shows what its host is doing', () => {
		370	expect(vmDetail(vmCreating('downloading image 1.2/3.7 GiB'))).toBe(
		371	'downloading image 1.2/3.7 GiB'
		372	);
		373	});
		374
		375	test('a host with nothing to say gets no row', () => {
		376	expect(vmDetail(vmCreating(''))).toBe('');
		377	});
		378
		379	test('a settled VM says nothing, even carrying a detail', () => {
		380	// The sentence describes work in flight. On a ready VM it would be the
		381	// past presented as the present.
		382	expect(vmDetail({ ...vmTrusting(null), status_detail: 'booting' })).toBe('');
		383	});
		384	});
		385
		386	describe('sessionSummary', () => {
		387	test('a counted port says what it holds and what it has turned away', () => {
		388	expect(sessionSummary(exposure({ active: 7, refused: 12, dropped: 3 }))).toBe(
		389	'7 open · 12 refused, 3 dropped since agent start'
		390	);
		391	});
		392
		393	test('a counted port with nothing to report still says so', () => {
		394	// Reported zeros are a fact: this port has turned nobody away.
		395	expect(sessionSummary(exposure({ active: 0, refused: 0, dropped: 0 }))).toBe(
		396	'0 open · 0 refused, 0 dropped since agent start'
		397	);
		398	});
		399
		400	test('a port nobody counted says nothing rather than zero', () => {
		401	// An older agent, or an exposure not yet reported on. Zeros here would
		402	// claim the port has refused nobody, which nothing on the wire supports.
		403	expect(sessionSummary(exposure(null))).toBe('');
		404	expect(sessionSummary(exposure(undefined))).toBe('');
		405	});
		406	});

web/src/routes/+page.svelte

Old		New
@@ -8,6 +8,7 @@
8	decommissionHost,	8	decommissionHost,
9	createJoinBlob,	9	createJoinBlob,
10	vmStatus,	10	vmStatus,
		11	vmDetail,
11	restoreVM,	12	restoreVM,
12	vmPower,	13	vmPower,
13	vmIP,	14	vmIP,
@@ -323,9 +324,11 @@
323	<td>{hostName(v.host_id)}</td>	324	<td>{hostName(v.host_id)}</td>
324	<td class="num">{v.vcpus}c · {v.mem_mb}MB · {v.disk_gb}GB</td>	325	<td class="num">{v.vcpus}c · {v.mem_mb}MB · {v.disk_gb}GB</td>
325	<td>	326	<td>
326	{#if v.deleted}<span class="teardown">deleting—restorable</span>{:else}{vmStatus(v)}{/if}{v.last_error	327	{#if v.deleted}<span class="teardown">deleting—restorable</span>{:else}{vmStatus(v)}{/if}{vmDetail(
327	? ` · ${v.last_error}`	328	v
328	: ''}	329	)
		330	? ` · ${vmDetail(v)}`
		331	: ''}{v.last_error ? ` · ${v.last_error}` : ''}
329	</td>	332	</td>
330	<td>{vmPower(v)}</td>	333	<td>{vmPower(v)}</td>
331	<td class="actions">	334	<td class="actions">

web/src/routes/vms/[id]/+page.svelte

Old		New
@@ -12,6 +12,8 @@
12	teardownApprox,	12	teardownApprox,
13	vmPower,	13	vmPower,
14	vmIP,	14	vmIP,
		15	vmDetail,
		16	sessionSummary,
15	vmPowerAction,	17	vmPowerAction,
16	deleteConfirm,	18	deleteConfirm,
17	vmEvents,	19	vmEvents,
@@ -236,7 +238,15 @@
236	<tbody>	238	<tbody>
237	<tr><th>ID</th><td>{vm.id}</td></tr>	239	<tr><th>ID</th><td>{vm.id}</td></tr>
238	<tr><th>Host</th><td>{#if host}<a href="/hosts/{host.id}">{host.name}</a>{:else}{vm.host_id}{/if}</td></tr>	240	<tr><th>Host</th><td>{#if host}<a href="/hosts/{host.id}">{host.name}</a>{:else}{vm.host_id}{/if}</td></tr>
239	<tr><th>Status</th><td>{vmStatus(vm)}</td></tr>	241	<tr>
		242	<th>Status</th>
		243	<td>
		244	{vmStatus(vm)}
		245	<!-- What the host is doing about this VM right now, when it is
		246	doing something and has said so (see vmDetail). -->
		247	{#if vmDetail(vm)}<span class="hint detail">{vmDetail(vm)}</span>{/if}
		248	</td>
		249	</tr>
240	<tr><th>Power</th><td>{vmPower(vm)} (desired: {vm.power_state})</td></tr>	250	<tr><th>Power</th><td>{vmPower(vm)} (desired: {vm.power_state})</td></tr>
241	<tr><th>IP</th><td>{vmIP(vm)} <span class="hint">(host bridge, NAT—not reachable off-host)</span></td></tr>	251	<tr><th>IP</th><td>{vmIP(vm)} <span class="hint">(host bridge, NAT—not reachable off-host)</span></td></tr>
242	<tr><th>Resources</th><td>{vm.vcpus}c / {vm.mem_mb}MB / {vm.disk_gb}GB</td></tr>	252	<tr><th>Resources</th><td>{vm.vcpus}c / {vm.mem_mb}MB / {vm.disk_gb}GB</td></tr>
@@ -338,6 +348,11 @@
338	{/if}	348	{/if}
339	<span class="state {exposureState(e)}">●</span> {exposureState(e)}	349	<span class="state {exposureState(e)}">●</span> {exposureState(e)}
340	{#if e.reason && !tearingDown}<span class="err">{e.reason}</span>{/if}	350	{#if e.reason && !tearingDown}<span class="err">{e.reason}</span>{/if}
		351	<!-- Only a port whose host actually counted gets this line; see
		352	sessionSummary. -->
		353	{#if sessionSummary(e) && !tearingDown}
		354	<span class="hint detail">{sessionSummary(e)}</span>
		355	{/if}
341	</td>	356	</td>
342	<td>	357	<td>
343	{#if !tearingDown}	358	{#if !tearingDown}
@@ -405,6 +420,13 @@
405	.err {	420	.err {
406	color: var(--bad);	421	color: var(--bad);
407	}	422	}
		423	/* A fact's quieter second half: what is happening under a status, what a
		424	published port has carried. It carries the global .hint colour and adds
		425	only the line of its own — it is a sentence, not a value, and reading it
		426	run together with the fact above it is what a line break prevents. */
		427	.detail {
		428	display: block;
		429	}
408	/* The trusted CAs are a list inside a value cell: unbulleted and flush, so	430	/* The trusted CAs are a list inside a value cell: unbulleted and flush, so
409	the row still reads as one fact with several parts rather than as a	431	the row still reads as one fact with several parts rather than as a
410	nested table. Usually one or two entries. */	432	nested table. Usually one or two entries. */