Let plans wait for a plan to move and answer it as data (issue 229)
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request

An agent following a rollout had plans answer once, in prose, and polled the
console with a curl loop of its own to be told a plan moved. plans now takes
wait (a plan's id), until (moved, tier or done) and bound, and answers when the
plan moves or the bound passes, saying which, with the plan as data and its
line as the output.

A call's one answer has to reach the console inside its thirty seconds, so
link.AnswersWithin lets a handler that waits on purpose hold its answer up to
LongestAnswer (25s) instead of AnswerWithin; the bound is at most 20s, and a
caller following a plan further asks again.
This commit is contained in:
jochen
2026-10-12 16:18:54 +02:00
parent 9eddf36503
commit d20b52cb19
8 changed files with 484 additions and 14 deletions
+47 -10
View File
@@ -601,6 +601,24 @@ func Acknowledge(ctx context.Context) {
}
}
// LongestAnswer is the longest a handler that waits on purpose may hold its caller's one answer
// (AnswersWithin, novox/hq issue 229): inside the console's thirty seconds with room for the trip back,
// and inside the bus's window for an answer (broker.ResponseTTL). A variable so a test need not wait.
var LongestAnswer = 25 * time.Second
type answerWithinKey struct{}
// AnswersWithin asks that the call's caller be answered in full if the handler ends within d of the
// call's start, rather than within AnswerWithin: for a handler that waits on purpose, with a bound —
// `plans` waiting for a plan to move (novox/hq issue 229). Past d, or past LongestAnswer whichever is
// shorter, its caller is told it is running, as for any call. Never shortens the window; without effect
// outside a served call, and once the caller was answered.
func AnswersWithin(ctx context.Context, d time.Duration) {
if f, ok := ctx.Value(answerWithinKey{}).(func(time.Duration)); ok {
f(d)
}
}
// running is the interim answer: what a caller reads when the call has not finished.
func running(c *Call, within time.Duration, acknowledged bool, follow string) []byte {
why := fmt.Sprintf("it has not finished in %s", within)
@@ -645,6 +663,13 @@ func (l *CallLog) serveCallWithin(seat, verb string, args json.RawMessage, reply
acknowledged := make(chan struct{})
var once sync.Once
ctx = context.WithValue(ctx, answerNowKey{}, func() { once.Do(func() { close(acknowledged) }) })
longer := make(chan time.Duration, 1)
ctx = context.WithValue(ctx, answerWithinKey{}, func(d time.Duration) {
select {
case longer <- d:
default:
}
})
type outcome struct {
body []byte
@@ -703,17 +728,29 @@ func (l *CallLog) serveCallWithin(seat, verb string, args json.RawMessage, reply
}
}
}
timer := time.NewTimer(AnswerWithin)
within, start := AnswerWithin, time.Now()
timer := time.NewTimer(within)
defer timer.Stop()
select {
case o := <-done:
l.finish(c, o.body, o.failed, false)
say(o.body)
return
case <-acknowledged:
say(running(c, AnswerWithin, true, l.Follow))
case <-timer.C:
say(running(c, AnswerWithin, false, l.Follow))
waiting:
for {
select {
case o := <-done:
l.finish(c, o.body, o.failed, false)
say(o.body)
return
case <-acknowledged:
say(running(c, within, true, l.Follow))
break waiting
case d := <-longer:
// A handler waiting on purpose (AnswersWithin): its window is stretched, never past LongestAnswer.
if d = min(d, LongestAnswer); d > within {
within = d
timer.Reset(time.Until(start.Add(d)))
}
case <-timer.C:
say(running(c, within, false, l.Follow))
break waiting
}
}
o := <-done
l.finish(c, o.body, o.failed, true)
+55
View File
@@ -0,0 +1,55 @@
package link
import (
"context"
"encoding/json"
"testing"
"time"
)
// **A handler that waits on purpose is answered in full when it ends inside the window it asked for**
// (novox/hq issue 229): `plans` asked to wait for a plan holds its caller's one answer until the plan
// moves or its bound passes, rather than being answered "still running" at AnswerWithin.
func TestAHandlerThatWaitsOnPurposeIsAnsweredInFullInsideItsWindow(t *testing.T) {
shortWindow(t, 20*time.Millisecond)
l, a := NewCallLog(), newAnswers(t)
l.serveCall("mesh-controller", "plans", json.RawMessage(`{"wait":"plan-1"}`), "_INBOX.x.3",
func(ctx context.Context, _ json.RawMessage) (any, error) {
AnswersWithin(ctx, 200*time.Millisecond)
time.Sleep(80 * time.Millisecond) // past AnswerWithin, inside what it asked for
return "plan-1 moved", nil
}, a.respond, nil)
if got := a.only()["result"]; got != "plan-1 moved" {
t.Fatalf("a waiting handler inside its window was answered %v, not in full", got)
}
}
// What a handler may ask for is bounded: past LongestAnswer its caller is told it is running, as any
// call is, so no answer is ever later than the console waits.
func TestAWaitIsNeverLongerThanTheLongestAnswer(t *testing.T) {
shortWindow(t, 10*time.Millisecond)
was := LongestAnswer
LongestAnswer = 30 * time.Millisecond
t.Cleanup(func() { LongestAnswer = was })
l, a := NewCallLog(), newAnswers(t)
release := make(chan struct{})
finished := make(chan struct{})
go func() {
l.serveCall("mesh-controller", "plans", nil, "_INBOX.x.4", func(ctx context.Context, _ json.RawMessage) (any, error) {
AnswersWithin(ctx, time.Hour)
<-release
return "late", nil
}, a.respond, nil)
close(finished)
}()
select {
case <-a.sent:
case <-time.After(2 * time.Second):
t.Fatal("a handler asking for an hour held its caller past LongestAnswer")
}
if got, _ := a.only()["result"].(map[string]any); got["running"] != true {
t.Fatalf("past LongestAnswer the caller was answered %v, not that it is running", a.only())
}
close(release)
<-finished
}