fix(harness): retry a failed revert step three times

2026-10-18 m2 배치에서 X-M2 복원 SSH 가 connection reset 으로 한 번 실패해 레거시
인스턴스에 netem loss 50% qdisc 가 남았고, 이후 14런의 V-G1·X-M2 주입이 막혔다.
복원 실패는 장애를 환경에 남기므로 15초 간격으로 최대 3회 다시 시도하고, 시도마다
이벤트를 남긴다. 3회 모두 실패하면 장애가 남아 있을 수 있다는 note 를 기록한다.

ai/cmd/agent/main.go

@@ -7,11 +7,14 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/agent"
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/anthropic"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/diag"
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/external"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/harness"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
@@ -19,6 +22,15 @@ import (
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/tools"
 )
 
+func envInt(k string, d int) int {
+	if v := os.Getenv(k); v != "" {
+		if n, err := strconv.Atoi(v); err == nil {
+			return n
+		}
+	}
+	return d
+}
+
 func env(k, d string) string {
 	if v := os.Getenv(k); v != "" {
 		return v
@@ -95,11 +107,38 @@ func main() {
 		HitlURL: env("HITL_URL", "http://hitl:8081"),
 	}
 	ag := agent.New(acfg, oai, tr, hc, store, enums, pv)
-	jevMode := env("JEV_MODE", "off") // off | shadow (shadow: Jev answers are recorded next to the LLM diagnosis, never acted on)
+	// Claude Messages API for claude-* model ids chosen per campaign; the key may be registered later (empty = diagnoses with
+	// a claude-* model fail with "anthropic api key is empty", recorded in the harness diagnosis event).
+	anth := anthropic.New(env("ANTHROPIC_BASE_URL", ""), keyReader(env("ANTHROPIC_KEY_FILE", "/run/secrets/anthropic-api-key")))
+	// Workload Identity Federation (preferred, no static key): the four ids are plain settings in ai.env; the VM's
+	// Google identity token is exchanged for a short-lived access token (internal/anthropic/federation.go).
+	anth.SetFederation(&anthropic.Federation{
+		RuleID: env("ANTHROPIC_FEDERATION_RULE_ID", ""), OrgID: env("ANTHROPIC_ORGANIZATION_ID", ""),
+		ServiceAccountID: env("ANTHROPIC_SERVICE_ACCOUNT_ID", ""), WorkspaceID: env("ANTHROPIC_WORKSPACE_ID", ""),
+		IdentityTokenURL: env("ANTHROPIC_IDENTITY_TOKEN_URL", ""),
+	})
+	ag.SetAnthropic(anth)
+	log.Printf("anthropic auth: %s", anth.AuthMode())
+	// external-* model ids (reference runs, 2026-10-16): each loop turn is served to an outside client over
+	// GET /external/next and answered through POST /external/reply (internal/external). The diagnosis deadline is
+	// EXTERNAL_DIAGNOSE_TIMEOUT_S (the harness waits a little longer for external-* models), one turn at most
+	// EXTERNAL_TURN_TIMEOUT_S. Off unless EXTERNAL_ADDR is set; the endpoints live on that separate listener, which
+	// the VM compose file maps to 127.0.0.1 only (reachable through the IAP SSH tunnel, never from the VPC): a reply
+	// is a diagnosis and, under auto-approval, an executed action.
+	var ext *external.Responder
+	if addr := env("EXTERNAL_ADDR", ""); addr != "" {
+		ext = external.New(time.Duration(envInt("EXTERNAL_TURN_TIMEOUT_S", 1200)) * time.Second)
+		ag.SetExternal(ext)
+		esrv := &http.Server{Addr: addr, Handler: logging(ext.Handler()), ReadHeaderTimeout: 10 * time.Second}
+		go func() { log.Fatal(esrv.ListenAndServe()) }()
+		log.Printf("external responder: listening on %s", addr)
+	}
+	externalDiagnoseTimeout := time.Duration(envInt("EXTERNAL_DIAGNOSE_TIMEOUT_S", 1500)) * time.Second
+	jevMode := env("JEV_MODE", "off") // off | shadow | enforce (shadow records Jev answers next to the diagnosis; enforce also applies the two review rules)
 	var jc *jev.Client
-	if jevMode == "shadow" {
+	if jevMode == "shadow" || jevMode == "enforce" {
 		jc = jev.New(env("JEV_BASE_URL", ""), env("JEV_MODEL", "jev-1.13.0"), keyReader(env("JEV_KEY_FILE", "/run/secrets/typesafe-api-key")))
-		ag.EnableShadow(jc)
+		ag.EnableShadow(jc, jevMode == "enforce")
 	}
 	log.Printf("agent up: model=%s prompt=%s labels=%d actions=%d jev=%s", acfg.Model, pv, len(enums.Labels), len(enums.Actions), jevMode)
 	if jc != nil && !jc.KeyPresent() {
@@ -130,6 +169,16 @@ func main() {
 		out := map[string]any{"model": acfg.Model, "prompt_version": pv, "labels": enums.Labels, "actions": enums.Actions}
 		key, kerr := keyReader(env("OPENAI_KEY_FILE", "/run/secrets/openai-api-key"))()
 		out["openai_key_present"] = kerr == nil && key != ""
+		out["anthropic_key_present"] = anth.KeyPresent()
+		out["anthropic_auth"] = anth.AuthMode()
+		if f := anth.Federation(); f != nil {
+			out["anthropic_federation"] = f.Probe(c, anth.BaseURL) // one real token exchange; the Console history page shows the deny reason on failure
+		}
+		if ext != nil {
+			out["external_responder"] = ext.Status()
+		} else {
+			out["external_responder"] = "off"
+		}
 		out["jev_mode"] = jevMode
 		if jc != nil {
 			out["jev_key_present"] = jc.KeyPresent()
@@ -162,7 +211,11 @@ func main() {
 		}
 		inflight++
 		defer func() { inflight-- }()
-		c, cancel := context.WithTimeout(context.Background(), 170*time.Second) // harness waits 3 minutes
+		timeout := 170 * time.Second // harness waits 3 minutes
+		if agent.IsExternal(req.Model) {
+			timeout = externalDiagnoseTimeout
+		}
+		c, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 		prefetched := ag.Prefetch(c)
 		tri := ag.ShadowTriage(c, req, prefetched) // nil unless JEV_MODE=shadow; recorded only
@@ -173,10 +226,11 @@ func main() {
 			return
 		}
 		rev := ag.ShadowReview(c, req, res)
-		ag.Dispose(c, req, res)
 		if tri != nil || rev != nil {
-			res.Jev = map[string]any{"mode": "shadow", "triage": tri, "review": rev}
+			res.Jev = map[string]any{"mode": jevMode, "triage": tri, "review": rev}
 		}
+		ag.Enforce(res, rev) // no-op unless JEV_MODE=enforce
+		ag.Dispose(c, req, res)
 		ag.SaveShadow(context.Background(), req, res.SessionID, res, tri, rev)
 		writeJSON(w, 200, res)
 	})

ai/cmd/hitl/main.go

@@ -59,6 +59,7 @@ func main() {
 		}
 		connected = true
 		bot.RearmPending(ctx)
+		go bot.SyncLoop(ctx, 60*time.Second)
 	}
 
 	mux := http.NewServeMux()

ai/internal/agent/agent.go

@@ -35,9 +35,17 @@ type Config struct {
 	ConfidenceMin   float64 // 0.5 (design 6.3)
 }
 
+// Responder is the one call the diagnosis loop needs from an LLM provider (OpenAI Responses API natively,
+// the Claude Messages API through internal/anthropic).
+type Responder interface {
+	CreateResponse(ctx context.Context, req openai.Request) (*openai.Response, error)
+}
+
 type Agent struct {
 	cfg        Config
 	oai        *openai.Client
+	anth       Responder // nil = claude-* models unavailable
+	ext        Responder // nil = external-* models unavailable (internal/external: a client outside the VM answers each turn)
 	tools      *tools.Runner
 	harness    *harness.Client
 	store      *rag.Store
@@ -45,6 +53,7 @@ type Agent struct {
 	promptV    string
 	http       *http.Client
 	jev        *jev.Client // nil = shadow gates off
+	jevEnforce bool        // JEV_MODE=enforce: review rules may downgrade a proposal to escalate
 }
 
 func New(cfg Config, oai *openai.Client, tr *tools.Runner, hc *harness.Client, store *rag.Store, enums diag.Enums, promptVersion string) *Agent {
@@ -63,14 +72,42 @@ func New(cfg Config, oai *openai.Client, tr *tools.Runner, hc *harness.Client, s
 	return &Agent{cfg: cfg, oai: oai, tools: tr, harness: hc, store: store, enums: enums, promptV: promptVersion, http: &http.Client{Timeout: 30 * time.Second}}
 }
 
-// EnableShadow turns the Jev shadow gates on (JEV_MODE=shadow). Answers are recorded only.
+// SetAnthropic enables claude-* model ids (Claude Messages API adapter).
-func (a *Agent) EnableShadow(c *jev.Client) { a.jev = c }
+func (a *Agent) SetAnthropic(c Responder) { a.anth = c }
+
+// SetExternal enables external-* model ids (turns served to an outside client over /external/next, reference runs).
+func (a *Agent) SetExternal(c Responder) { a.ext = c }
+
+// IsExternal reports whether the model id is served by the external responder (longer deadlines apply).
+func IsExternal(model string) bool { return strings.HasPrefix(model, "external-") }
+
+// provider picks the LLM client for a model id: claude-* -> Anthropic adapter, external-* -> external responder,
+// everything else -> OpenAI.
+func (a *Agent) provider(model string) (Responder, error) {
+	if strings.HasPrefix(model, "claude-") {
+		if a.anth == nil {
+			return nil, fmt.Errorf("model %s: anthropic provider not configured", model)
+		}
+		return a.anth, nil
+	}
+	if IsExternal(model) {
+		if a.ext == nil {
+			return nil, fmt.Errorf("model %s: external responder not configured", model)
+		}
+		return a.ext, nil
+	}
+	return a.oai, nil
+}
+
+// EnableShadow turns the Jev gates on (JEV_MODE=shadow records only; enforce also applies the review rules).
+func (a *Agent) EnableShadow(c *jev.Client, enforce bool) { a.jev, a.jevEnforce = c, enforce }
 
 // Request is the harness -> AI VM contract (harness alerts.go diagnose()).
 type Request struct {
 	Alert      map[string]any `json:"alert"`
 	RunID      *string        `json:"run_id"`
 	ScenarioID *string        `json:"scenario_id"`
+	Model      string         `json:"model,omitempty"` // campaign-level model selection (harness runs.model); "" = Config.Model
 }
 
 // Result is returned to the harness: the diagnosis fields at top level (the harness stores the whole object as the
@@ -83,6 +120,7 @@ type Result struct {
 	Turns         int            `json:"turns"`
 	ToolCalls     int            `json:"tool_calls"`
 	DurationMs    int64          `json:"duration_ms"`
+	Usage         map[string]int `json:"usage,omitempty"` // input_tokens / output_tokens / total_tokens summed over the turns
 	Disposition   string         `json:"disposition"`     // proposal | escalated | none
 	ProposalToken string         `json:"proposal_token,omitempty"`
 	Jev           map[string]any `json:"jev,omitempty"` // shadow gate answers (recorded, never acted on)
@@ -154,13 +192,17 @@ func (a *Agent) Diagnose(ctx context.Context, req Request) (*Result, error) {
 func (a *Agent) DiagnoseWith(ctx context.Context, req Request, prefetched string) (*Result, error) {
 	start := time.Now()
 	sid := "ses-" + uuid.NewString()[:8]
-	res := &Result{SessionID: sid, Model: a.cfg.Model, PromptVersion: a.promptV}
+	model := a.cfg.Model
+	if req.Model != "" {
+		model = req.Model
+	}
+	res := &Result{SessionID: sid, Model: model, PromptVersion: a.promptV}
 	var calls []toolCallLog
 	input := []openai.Item{
 		{Type: "message", Role: "system", Content: a.cfg.SystemPrompt},
 		{Type: "message", Role: "user", Content: a.userMessage(ctx, req, prefetched)},
 	}
-	session := rag.Session{SessionID: sid, RunID: req.RunID, ScenarioID: req.ScenarioID, Model: a.cfg.Model, PromptVersion: a.promptV, Input: map[string]any{"alert": req.Alert, "user_message": input[1].Content}}
+	session := rag.Session{SessionID: sid, RunID: req.RunID, ScenarioID: req.ScenarioID, Model: model, PromptVersion: a.promptV, Input: map[string]any{"alert": req.Alert, "user_message": input[1].Content}}
 	finish := func(err error) {
 		session.Output = res
 		session.ToolCalls = calls
@@ -175,6 +217,11 @@ func (a *Agent) DiagnoseWith(ctx context.Context, req Request, prefetched string
 		}
 	}
 
+	llm, err := a.provider(model)
+	if err != nil {
+		finish(err)
+		return nil, err
+	}
 	defs := a.toolDefs()
 	var prevID string
 	var diagnosis *diag.Diagnosis
@@ -184,14 +231,14 @@ func (a *Agent) DiagnoseWith(ctx context.Context, req Request, prefetched string
 	usage := map[string]int{"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "turns": 0}
 	for turn := 1; turn <= a.cfg.MaxTurns && diagnosis == nil; turn++ {
 		res.Turns = turn
-		r := openai.Request{Model: a.cfg.Model, Input: input, Tools: defs, ToolChoice: "auto", MaxOutputTokens: a.cfg.MaxOutputTokens, ParallelToolCalls: &pf}
+		r := openai.Request{Model: model, Input: input, Tools: defs, ToolChoice: "auto", MaxOutputTokens: a.cfg.MaxOutputTokens, ParallelToolCalls: &pf}
 		if prevID != "" {
 			r.PreviousResponseID = prevID
 		}
 		if turn == a.cfg.MaxTurns { // last chance: force the final call
 			r.ToolChoice = map[string]any{"type": "function", "name": "submit_diagnosis"}
 		}
-		resp, err := a.oai.CreateResponse(ctx, r)
+		resp, err := llm.CreateResponse(ctx, r)
 		if err != nil {
 			finish(err)
 			return nil, err
@@ -227,6 +274,7 @@ func (a *Agent) DiagnoseWith(ctx context.Context, req Request, prefetched string
 		}
 	}
 	session.Usage = usage
+	res.Usage = usage
 	res.ToolCalls = len(calls)
 	res.DurationMs = time.Since(start).Milliseconds()
 	if diagnosis == nil {

ai/internal/agent/agent_test.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/anthropic"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/diag"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/harness"
 	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
@@ -81,3 +82,70 @@ func TestTargetFor(t *testing.T) {
 		t.Fatal("mig")
 	}
 }
+
+// A claude-* model id routes the same loop through the Anthropic adapter; the request model wins over Config.Model
+// and the summed usage is returned in the result.
+func TestDiagnoseRoutesClaudeModel(t *testing.T) {
+	var models []string
+	anth := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var b map[string]any
+		_ = json.NewDecoder(r.Body).Decode(&b)
+		models = append(models, b["model"].(string))
+		args := `{"root_cause_label":"db_primary_down","confidence":0.8,"evidence":["e"],"proposed_action":"escalate","target":"db-primary","rationale":"r"}`
+		w.Write([]byte(`{"id":"msg_a","type":"message","model":"claude-sonnet-5-5","stop_reason":"tool_use","content":[{"type":"tool_use","id":"toolu_a","name":"submit_diagnosis","input":` + args + `}],"usage":{"input_tokens":40,"output_tokens":8}}`))
+	}))
+	defer anth.Close()
+	oaiSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/embeddings" {
+			w.Write([]byte(`{"data":[{"index":0,"embedding":[0.1,0.2]}]}`))
+			return
+		}
+		t.Errorf("openai must not be called for a claude-* model: %s", r.URL.Path)
+	}))
+	defer oaiSrv.Close()
+	mimir := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte(`{"status":"success","data":{"resultType":"vector","result":[]}}`))
+	}))
+	defer mimir.Close()
+	oai := openai.New(oaiSrv.URL, func() (string, error) { return "test-key", nil })
+	tr := tools.New(tools.Config{MimirURL: mimir.URL, LokiURL: mimir.URL, EmbedModel: "e"}, nil, oai)
+	enums := diag.Enums{Labels: diag.DefaultLabels, Actions: diag.DefaultActions}
+	ag := New(Config{Model: "gpt-default", SystemPrompt: "sys", Topology: "topo", SLO: "slo", HitlURL: "http://127.0.0.1:1", MaxTurns: 3}, oai, tr, harness.New("http://127.0.0.1:1"), nil, enums, "pv1")
+	req := Request{Alert: map[string]any{"alertname": "DbPrimaryDown"}, Model: "claude-sonnet-5-5"}
+	if _, err := ag.Diagnose(context.Background(), req); err == nil || !strings.Contains(err.Error(), "anthropic provider not configured") {
+		t.Fatalf("without SetAnthropic a claude-* model must fail clearly, got %v", err)
+	}
+	ag.SetAnthropic(anthropic.New(anth.URL, func() (string, error) { return "k", nil }))
+	res, err := ag.Diagnose(context.Background(), req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Model != "claude-sonnet-5-5" || len(models) != 1 || models[0] != "claude-sonnet-5-5" || res.RootCauseLabel != "db_primary_down" {
+		t.Fatalf("routing: res=%+v models=%v", res, models)
+	}
+	if res.Usage["input_tokens"] != 40 || res.Usage["output_tokens"] != 8 || res.Usage["total_tokens"] != 48 || res.Usage["turns"] != 1 {
+		t.Fatalf("usage: %v", res.Usage)
+	}
+}
+
+// Model routing: external-* goes to the external responder when one is set, and is refused when none is configured.
+func TestProviderRouting(t *testing.T) {
+	a := New(Config{}, nil, nil, nil, nil, diag.Enums{}, "v")
+	if _, err := a.provider("external-claude-mythos-5-1"); err == nil || !strings.Contains(err.Error(), "external responder not configured") {
+		t.Fatalf("expected unconfigured error, got %v", err)
+	}
+	fake := &fakeResponder{}
+	a.SetExternal(fake)
+	if p, err := a.provider("external-claude-mythos-5-1"); err != nil || p != Responder(fake) {
+		t.Fatalf("external routing: %v %v", p, err)
+	}
+	if !IsExternal("external-x") || IsExternal("claude-x") || IsExternal("gpt-6-luna") {
+		t.Fatal("IsExternal prefix check")
+	}
+}
+
+type fakeResponder struct{}
+
+func (*fakeResponder) CreateResponse(context.Context, openai.Request) (*openai.Response, error) {
+	return nil, nil
+}

ai/internal/agent/enforce_test.go

@@ -0,0 +1,50 @@
+package agent
+
+import (
+	"testing"
+
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev"
+)
+
+func f(v float64) *float64 { return &v }
+
+func TestEnforceRules(t *testing.T) {
+	a := &Agent{jevEnforce: true}
+	// rule A: best_action disagrees with high confidence -> escalate
+	res := &Result{}
+	res.ProposedAction, res.Target = "net.route_change", ""
+	rev := &ShadowRecord{Gate: "review", Answers: map[string]any{"best_action": jev.Answer{Type: "choice", Choice: "escalate", Confidence: f(0.95)}}}
+	a.Enforce(res, rev)
+	if res.ProposedAction != "escalate" || res.Jev["enforced"].(map[string]any)["rule"] != "best_action_disagrees" {
+		t.Fatalf("rule A: %+v %v", res.Diagnosis, res.Jev)
+	}
+	// rule B: target invalid -> escalate
+	res = &Result{}
+	res.ProposedAction, res.Target = "k8s.rollout_restart", ""
+	rev = &ShadowRecord{Gate: "review", Answers: map[string]any{"best_action": jev.Answer{Type: "choice", Choice: "k8s.rollout_restart", Confidence: f(0.9)}, "target_valid": jev.Answer{Type: "noul", Noul: f(0.2)}}}
+	a.Enforce(res, rev)
+	if res.ProposedAction != "escalate" || res.Jev["enforced"].(map[string]any)["rule"] != "target_invalid" {
+		t.Fatalf("rule B: %+v %v", res.Diagnosis, res.Jev)
+	}
+	// agreement with valid target -> unchanged; none/escalate -> untouched; shadow mode -> untouched
+	res = &Result{}
+	res.ProposedAction, res.Target = "k8s.rollout_restart", "api"
+	rev = &ShadowRecord{Gate: "review", Answers: map[string]any{"best_action": jev.Answer{Choice: "k8s.rollout_restart", Confidence: f(0.99)}, "target_valid": jev.Answer{Noul: f(0.9)}}}
+	a.Enforce(res, rev)
+	if res.ProposedAction != "k8s.rollout_restart" || res.Jev != nil {
+		t.Fatalf("agreement: %+v", res.Diagnosis)
+	}
+	res = &Result{}
+	res.ProposedAction = "none"
+	a.Enforce(res, &ShadowRecord{Answers: map[string]any{"best_action": jev.Answer{Choice: "escalate", Confidence: f(0.99)}}})
+	if res.ProposedAction != "none" {
+		t.Fatalf("none must not change")
+	}
+	s := &Agent{jevEnforce: false}
+	res = &Result{}
+	res.ProposedAction = "k8s.scale"
+	s.Enforce(res, &ShadowRecord{Answers: map[string]any{"best_action": jev.Answer{Choice: "escalate", Confidence: f(0.99)}}})
+	if res.ProposedAction != "k8s.scale" {
+		t.Fatalf("shadow must not change")
+	}
+}

ai/internal/agent/shadow.go

@@ -78,25 +78,90 @@ func (a *Agent) ShadowTriage(ctx context.Context, req Request, prefetched string
 	rec := &ShadowRecord{Gate: "triage"}
 	rec.state = map[string]any{
 		"alert":             compactAlert(req.Alert),
+		"now":               a.nowSummary(ctx, req.Alert),
 		"samples_last_10m":  prefetched,
+		"recent_error_logs": a.recentErrorLogs(ctx),
 		"slo":               "availability: probe_success = 1; latency: api p95 < 0.5 s; errors: 5xx ratio < 1%",
 	}
 	rec.questions = map[string]jev.Question{
-		"label": {Type: "choice", Instructions: "Which root cause label best explains `alert` together with `samples_last_10m`?", Criteria: labels},
+		"label": {Type: "choice", Instructions: "Which root cause label best explains `alert`, `now`, `samples_last_10m` and `recent_error_logs`?", Criteria: labels},
-		"recovering": {Type: "noul", Instructions: "Looking only at the newest values in `samples_last_10m`, is the symptom already gone: probe_success back to 1 and api_p95_latency_s and api_error_ratio inside `slo`?",
+		"recovering": {Type: "noul", Instructions: "Look only at `now`: is `now.recovered_now` true, meaning the alert's own condition no longer holds (`now.alert_condition_clear`) and the newest probe, the 1-minute p95 and the 1-minute error ratio are all inside `slo` right now?",
-			Criteria: map[string]string{"true": "the newest samples are inside the SLO again", "false": "the newest samples still show the failure, or the samples are missing"}},
+			Criteria: map[string]string{"true": "now.recovered_now is true: the alert condition is clear and the SLO values are inside", "false": "now.recovered_now is false, or the values are missing"}},
-		"needs_investigation": {Type: "noul", Instructions: "To choose a safe remediation action, would an operator need evidence that is not in this state (pod logs, Kubernetes objects, VM status, database replication status)?",
+		"needs_investigation": {Type: "noul", Instructions: "To choose a safe remediation action, would an operator need evidence that is not in this state (more logs, Kubernetes objects, VM status, database replication status)?",
 			Criteria: map[string]string{"true": "more evidence is needed before any action", "false": "this state alone is enough to decide the action"}},
-		"layer": {Type: "choice", Instructions: "Which recovery layer handles this alert?", Criteria: map[string]string{
+		"irreversible_needed": {Type: "noul", Instructions: "Would fixing the root cause shown in this state require an irreversible or human-only operation: promoting the database replica, repairing a database schema or data, or starting or recreating a stopped VM?",
-			"layer1": "the platform recovers by itself (ReplicaSet, Argo CD selfHeal, MIG autohealing, compose restart); no action needed",
+			Criteria: map[string]string{"true": "the fix is irreversible or human-only (database promotion, schema or data repair, VM lifecycle)", "false": "the platform recovers by itself or a reversible catalogue action is enough"}},
-			"layer2": "a reversible catalogue action with human approval is needed",
-			"layer3": "an irreversible or human-only action is needed (database promotion, data fix, instance lifecycle)",
-		}},
 	}
 	a.ask(ctx, rec)
 	return rec
 }
 
+// nowSummary computes the newest values and the SLO verdict in code (Jev reads text literally and should not compare
+// numbers or read the tail of a series: jev-1.13 jaggedness notes).
+func (a *Agent) nowSummary(ctx context.Context, alert map[string]any) map[string]any {
+	q := func(promql string) any {
+		v, err := a.tools.Instant(ctx, promql)
+		if err != nil || v != v { // NaN
+			return nil
+		}
+		return v
+	}
+	probeALB := q(`probe_success{target="alb"}`)
+	probeDB := q(`probe_success{target="db-primary"}`)
+	probeMin1m := q(`min_over_time(probe_success{target="alb"}[1m])`)
+	probeMin3m := q(`min_over_time(probe_success{target="alb"}[3m])`)
+	p95 := q(`histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket{route=~"orders|price"}[1m])))`)
+	errRatio := q(`sum(rate(demo_http_requests_total{route=~"orders|price",status=~"5.."}[1m])) / sum(rate(demo_http_requests_total{route=~"orders|price"}[1m]))`)
+	// the alert's own condition (Grafana annotation `condition`, the rule expression in bool form). A fault that does
+	// not touch the SLO (replay paused, memory pressure) is "recovering" only when this clears (campaign 3/4: 0.98).
+	var condNow any
+	condClear := true
+	if cond := alertCondition(alert); cond != "" {
+		condNow = q("max(" + cond + ")")
+		if v, ok := condNow.(float64); ok {
+			condClear = v == 0
+		}
+	}
+	restarts := q(`sum(increase(kube_pod_container_status_restarts_total{namespace="demo"}[10m]))`)
+	legacyUp := q(`count(up{job="legacy-api"} == 1)`)
+	f := func(v any) (float64, bool) {
+		x, ok := v.(float64)
+		return x, ok
+	}
+	pa, okA := f(probeALB)
+	pv, okP := f(p95)
+	ev, okE := f(errRatio)
+	probeOK := okA && pa >= 1
+	latOK := !okP || pv < 0.5 // no traffic = inside SLO (same rule as the harness judge)
+	errOK := !okE || ev < 0.01
+	return map[string]any{
+		"probe_alb_now": probeALB, "probe_db_primary_now": probeDB, "probe_alb_min_1m": probeMin1m, "probe_alb_min_3m": probeMin3m,
+		"api_p95_1m_s": p95, "api_error_ratio_1m": errRatio, "api_restarts_10m": restarts, "legacy_instances_up": legacyUp,
+		"alert_condition_now": condNow, "alert_condition_clear": condClear,
+		"probe_ok_now": probeOK, "latency_ok_now": latOK, "errors_ok_now": errOK, "slo_now_ok": probeOK && latOK && errOK,
+		"recovered_now": probeOK && latOK && errOK && condClear,
+	}
+}
+
+// alertCondition returns the rule expression carried in the alert annotations (empty when absent).
+func alertCondition(alert map[string]any) string {
+	ann, _ := alert["annotations"].(map[string]any)
+	if ann == nil {
+		return ""
+	}
+	c, _ := ann["condition"].(string)
+	return c
+}
+
+// recentErrorLogs returns a short, newest-first excerpt of api error lines (5 minutes, 8 lines) for the triage state.
+func (a *Agent) recentErrorLogs(ctx context.Context) string {
+	out := a.tools.Call(ctx, "query_logs", `{"logql":"{namespace=\"demo\", app=\"api\"} |~ \"(?i)error|timeout|refused|SQLSTATE\"","range_minutes":5,"limit":8}`)
+	if len(out) > 1800 {
+		out = out[:1800] + "..."
+	}
+	return out
+}
+
 // ShadowReview asks Jev to check the finished diagnosis against its evidence and the runbook of the proposed action.
 // It runs for every diagnosis (also none / escalate) so the comparison covers the whole campaign.
 func (a *Agent) ShadowReview(ctx context.Context, req Request, res *Result) *ShadowRecord {
@@ -127,19 +192,53 @@ func (a *Agent) ShadowReview(ctx context.Context, req Request, res *Result) *Sha
 			"rationale": d.Rationale, "proposed_action": d.ProposedAction, "target": d.Target, "harness_target": targetFor(d.ProposedAction, d.Target)},
 		"runbook": runbook,
 	}
+	catalogue := d.ProposedAction != "" && d.ProposedAction != "none" && d.ProposedAction != "escalate"
 	rec.questions = map[string]jev.Question{
 		"label_supported": {Type: "noul", Instructions: "Do the items in `diagnosis.evidence` support `diagnosis.root_cause_label`?",
 			Criteria: map[string]string{"true": "the evidence items show the condition that the label describes", "false": "the evidence is about something else, or too thin to support the label"}},
 		"runbook_conditions_met": {Type: "noul", Instructions: "Does `diagnosis.evidence` satisfy the conditions written after 'When' or '使う場面' in `runbook` for `diagnosis.proposed_action`? If `runbook` is empty, answer false.",
 			Criteria: map[string]string{"true": "each condition in the runbook is backed by a quoted evidence item", "false": "a condition is not backed by evidence, the evidence contradicts it, or the runbook is empty"}},
-		"target_valid": {Type: "noul", Instructions: "Is `diagnosis.target` a concrete, non-empty object name that the harness can act on for `diagnosis.proposed_action` (a deployment such as api or demo/api, a node name such as ip-10-10-10-11, a VM name such as selfheal-db-replica, or a legacy instance name such as legacy-xxxx)? For proposed_action none or escalate, answer true.",
-			Criteria: map[string]string{"true": "the target names one existing object of the kind the action needs", "false": "the target is empty, generic, or of the wrong kind"}},
 		"best_action": {Type: "choice", Instructions: "Given `diagnosis.evidence`, `diagnosis.rationale` and `runbook`, which catalogue action should be proposed? Prefer the smallest action; choose escalate for irreversible changes or weak evidence; choose none when the platform already recovers.", Criteria: actions},
 	}
+	if catalogue {
+		rec.questions["target_valid"] = jev.Question{Type: "noul", Instructions: "Is `diagnosis.target` a concrete, non-empty object name that the harness can act on for `diagnosis.proposed_action` (a deployment such as api or demo/api, a node name such as ip-10-10-10-11, a VM name such as selfheal-db-replica, or a legacy instance name such as legacy-xxxx)?",
+			Criteria: map[string]string{"true": "the target names one existing object of the kind the action needs", "false": "the target is empty, generic, or of the wrong kind"}}
+	}
 	a.ask(ctx, rec)
 	return rec
 }
 
+// Enforce applies the two review rules when JEV_MODE=enforce (2026-10-08 campaign-2 report, 3.3). It only ever downgrades
+// a catalogue proposal to escalate; none and escalate are never changed. The decision is recorded in res.Jev["enforced"].
+func (a *Agent) Enforce(res *Result, rev *ShadowRecord) {
+	if !a.jevEnforce || res == nil || rev == nil || rev.Answers == nil {
+		return
+	}
+	d := &res.Diagnosis
+	if d.ProposedAction == "" || d.ProposedAction == "none" || d.ProposedAction == "escalate" {
+		return
+	}
+	var rule, reason string
+	if ans, ok := rev.Answers["best_action"].(jev.Answer); ok && ans.Choice != "" && ans.Choice != d.ProposedAction && ans.Confidence != nil && *ans.Confidence >= 0.9 {
+		rule, reason = "best_action_disagrees", fmt.Sprintf("jev best_action=%s (confidence %.2f) != proposed %s", ans.Choice, *ans.Confidence, d.ProposedAction)
+	}
+	if rule == "" {
+		if ans, ok := rev.Answers["target_valid"].(jev.Answer); ok && ans.Noul != nil && *ans.Noul < 0.5 {
+			rule, reason = "target_invalid", fmt.Sprintf("jev target_valid=%.2f for target %q", *ans.Noul, d.Target)
+		}
+	}
+	if rule == "" {
+		return
+	}
+	log.Printf("jev enforce: %s: %s -> escalate", rule, reason)
+	if res.Jev == nil {
+		res.Jev = map[string]any{}
+	}
+	res.Jev["enforced"] = map[string]any{"rule": rule, "reason": reason, "from": d.ProposedAction, "to": "escalate"}
+	d.Rationale = fmt.Sprintf("%s [jev %s: %s]", d.Rationale, rule, reason)
+	d.ProposedAction = "escalate"
+}
+
 func (a *Agent) ask(ctx context.Context, rec *ShadowRecord) {
 	c, cancel := context.WithTimeout(ctx, 12*time.Second)
 	defer cancel()

ai/internal/anthropic/client.go

@@ -0,0 +1,321 @@
+// Package anthropic adapts the Claude Messages API to the small Responses-API surface the agent loop uses
+// (openai.Request / openai.Response with message, function_call and function_call_output items), so one loop
+// serves both providers. The agent picks this client for model ids starting with "claude-".
+//
+// Wire format (Messages API reference, anthropic-version 2023-06-01): system prompt as the top-level `system`
+// field, tools as {name, description, input_schema, strict}, tool calls as `tool_use` content blocks answered by
+// `tool_result` blocks in the next user message. The Responses-API `previous_response_id` chain is emulated
+// client-side: every response id maps to the full message list up to and including that assistant turn.
+//
+// Claude Opus 5.5 / Sonnet 5.5 reject forced tool use (tool_choice type "tool"/"any" -> 400), so a forced
+// tool_choice from the loop becomes {type:"auto"} plus a one-line instruction in the user turn.
+package anthropic
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
+)
+
+const (
+	apiVersion    = "2023-06-01"
+	maxConvCached = 64 // response ids kept for previous_response_id chaining (one diagnosis uses <= MaxTurns)
+	defaultTokens = 4000
+)
+
+type Client struct {
+	BaseURL string
+	HTTP    *http.Client
+	keyFn   func() (string, error)
+	fed     *Federation // nil = API key from keyFn; set = Workload Identity Federation (takes precedence)
+
+	mu    sync.Mutex
+	conv  map[string]conversation // response id -> messages (and system prompt) up to and including that assistant turn
+	order []string                // insertion order for eviction
+	seq   int
+}
+
+type conversation struct {
+	system string
+	msgs   []message
+}
+
+func New(baseURL string, keyFn func() (string, error)) *Client {
+	if baseURL == "" {
+		baseURL = "https://api.anthropic.com"
+	}
+	return &Client{BaseURL: strings.TrimRight(baseURL, "/"), HTTP: &http.Client{Timeout: 150 * time.Second}, keyFn: keyFn, conv: map[string]conversation{}}
+}
+
+// KeyPresent reports whether the key file has a non-empty value (the secret may not be registered yet).
+func (c *Client) KeyPresent() bool {
+	k, err := c.keyFn()
+	return err == nil && k != ""
+}
+
+// SetFederation switches the client to Workload Identity Federation (ignored when the ids are incomplete).
+func (c *Client) SetFederation(f *Federation) {
+	if f.Configured() {
+		c.fed = f
+	}
+}
+
+// AuthMode is "federation", "api_key" or "none" (reported by /selfcheck).
+func (c *Client) AuthMode() string {
+	if c.fed != nil {
+		return "federation"
+	}
+	if c.KeyPresent() {
+		return "api_key"
+	}
+	return "none"
+}
+
+// Federation returns the active federation config (nil when the API key path is used).
+func (c *Client) Federation() *Federation { return c.fed }
+
+// authorize sets the credential header: a federated bearer token, or the API key.
+func (c *Client) authorize(ctx context.Context, req *http.Request) error {
+	if c.fed != nil {
+		tok, err := c.fed.AccessToken(ctx, c.BaseURL)
+		if err != nil {
+			return fmt.Errorf("anthropic federation: %w", err)
+		}
+		req.Header.Set("Authorization", "Bearer "+tok)
+		return nil
+	}
+	key, err := c.keyFn()
+	if err != nil {
+		return fmt.Errorf("anthropic key: %w", err)
+	}
+	if key == "" {
+		return fmt.Errorf("anthropic api key is empty (secret selfheal-anthropic-api-key not registered yet) and federation is not configured")
+	}
+	req.Header.Set("x-api-key", key)
+	return nil
+}
+
+type message struct {
+	Role    string `json:"role"`
+	Content any    `json:"content"` // string or []block
+}
+
+type block struct {
+	Type      string          `json:"type"`
+	Text      string          `json:"text,omitempty"`
+	ID        string          `json:"id,omitempty"`          // tool_use
+	Name      string          `json:"name,omitempty"`        // tool_use
+	Input     json.RawMessage `json:"input,omitempty"`       // tool_use
+	ToolUseID string          `json:"tool_use_id,omitempty"` // tool_result
+	Content   string          `json:"content,omitempty"`     // tool_result
+}
+
+type tool struct {
+	Name        string         `json:"name"`
+	Description string         `json:"description,omitempty"`
+	InputSchema map[string]any `json:"input_schema"`
+	Strict      bool           `json:"strict,omitempty"`
+}
+
+type request struct {
+	Model      string         `json:"model"`
+	MaxTokens  int            `json:"max_tokens"`
+	System     string         `json:"system,omitempty"`
+	Messages   []message      `json:"messages"`
+	Tools      []tool         `json:"tools,omitempty"`
+	ToolChoice map[string]any `json:"tool_choice,omitempty"`
+}
+
+type response struct {
+	ID         string          `json:"id"`
+	Type       string          `json:"type"`
+	Model      string          `json:"model"`
+	StopReason string          `json:"stop_reason"`
+	Content    json.RawMessage `json:"content"` // kept verbatim: thinking blocks (with their signature) must be replayed unchanged
+	Usage      struct {
+		InputTokens              int `json:"input_tokens"`
+		OutputTokens             int `json:"output_tokens"`
+		CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
+		CacheReadInputTokens     int `json:"cache_read_input_tokens"`
+	} `json:"usage"`
+	Error *struct {
+		Type    string `json:"type"`
+		Message string `json:"message"`
+	} `json:"error"`
+}
+
+// forcedNudge is appended to the user turn when the loop asked for a forced submit_diagnosis call.
+const forcedNudge = "You must call the submit_diagnosis tool now with your final answer; do not call any other tool."
+
+// CreateResponse translates one loop turn to a Messages API call and the answer back to Responses-API items.
+func (c *Client) CreateResponse(ctx context.Context, req openai.Request) (*openai.Response, error) {
+	if c.fed == nil {
+		if key, err := c.keyFn(); err != nil {
+			return nil, fmt.Errorf("anthropic key: %w", err)
+		} else if key == "" {
+			return nil, fmt.Errorf("anthropic api key is empty (secret selfheal-anthropic-api-key not registered yet) and federation is not configured")
+		}
+	}
+	var msgs []message
+	var system []string
+	if req.PreviousResponseID != "" {
+		c.mu.Lock()
+		prev, ok := c.conv[req.PreviousResponseID]
+		c.mu.Unlock()
+		if !ok {
+			return nil, fmt.Errorf("anthropic: unknown previous_response_id %q", req.PreviousResponseID)
+		}
+		msgs = append(msgs, prev.msgs...)
+		if prev.system != "" {
+			system = append(system, prev.system) // the loop sends the system message only on the first turn
+		}
+	}
+	var userBlocks []block
+	flush := func() {
+		if len(userBlocks) == 0 {
+			return
+		}
+		msgs = append(msgs, message{Role: "user", Content: userBlocks})
+		userBlocks = nil
+	}
+	for _, it := range req.Input {
+		switch it.Type {
+		case "message":
+			text, _ := it.Content.(string)
+			switch it.Role {
+			case "system":
+				system = append(system, text)
+			case "assistant":
+				flush()
+				msgs = append(msgs, message{Role: "assistant", Content: []block{{Type: "text", Text: text}}})
+			default:
+				userBlocks = append(userBlocks, block{Type: "text", Text: text})
+			}
+		case "function_call_output":
+			userBlocks = append(userBlocks, block{Type: "tool_result", ToolUseID: it.CallID, Content: it.Output})
+		}
+	}
+	forced := false
+	if m, ok := req.ToolChoice.(map[string]any); ok && m["type"] == "function" {
+		forced = true
+	}
+	if forced {
+		userBlocks = append(userBlocks, block{Type: "text", Text: forcedNudge})
+	}
+	flush()
+	if len(msgs) == 0 {
+		return nil, fmt.Errorf("anthropic: no messages to send")
+	}
+	body := request{Model: req.Model, MaxTokens: req.MaxOutputTokens, System: strings.Join(system, "\n\n"), Messages: msgs}
+	if body.MaxTokens <= 0 {
+		body.MaxTokens = defaultTokens
+	}
+	for _, t := range req.Tools {
+		body.Tools = append(body.Tools, tool{Name: t.Name, Description: t.Description, InputSchema: t.Parameters, Strict: t.Strict})
+	}
+	if len(body.Tools) > 0 {
+		tc := map[string]any{"type": "auto"}
+		if req.ParallelToolCalls != nil && !*req.ParallelToolCalls {
+			tc["disable_parallel_tool_use"] = true
+		}
+		body.ToolChoice = tc
+	}
+
+	b, _ := json.Marshal(body)
+	send := func() (*http.Response, error) {
+		hreq, err := http.NewRequestWithContext(ctx, "POST", c.BaseURL+"/v1/messages", bytes.NewReader(b))
+		if err != nil {
+			return nil, err
+		}
+		if err := c.authorize(ctx, hreq); err != nil {
+			return nil, err
+		}
+		hreq.Header.Set("anthropic-version", apiVersion)
+		hreq.Header.Set("Content-Type", "application/json")
+		return c.HTTP.Do(hreq)
+	}
+	resp, err := send()
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode == 401 && c.fed != nil { // token revoked/expired early: exchange once more
+		resp.Body.Close()
+		c.fed.Invalidate()
+		if resp, err = send(); err != nil {
+			return nil, err
+		}
+	}
+	defer resp.Body.Close()
+	rb, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+	var r response
+	_ = json.Unmarshal(rb, &r)
+	if resp.StatusCode/100 != 2 {
+		msg := truncate(string(rb), 600)
+		if r.Error != nil {
+			msg = r.Error.Type + ": " + r.Error.Message
+		}
+		return nil, fmt.Errorf("anthropic /v1/messages: status %d: %s", resp.StatusCode, msg)
+	}
+	if r.Error != nil {
+		return nil, fmt.Errorf("anthropic response error: %s", r.Error.Message)
+	}
+
+	out := &openai.Response{ID: r.ID, Status: "completed", Model: r.Model}
+	if out.ID == "" {
+		c.mu.Lock()
+		c.seq++
+		out.ID = fmt.Sprintf("msg_local_%d", c.seq)
+		c.mu.Unlock()
+	}
+	var blocks []block
+	_ = json.Unmarshal(r.Content, &blocks)
+	var text strings.Builder
+	for _, bl := range blocks {
+		switch bl.Type {
+		case "text":
+			text.WriteString(bl.Text)
+		case "tool_use":
+			args := string(bl.Input)
+			if args == "" {
+				args = "{}"
+			}
+			out.Output = append(out.Output, openai.Item{Type: "function_call", CallID: bl.ID, Name: bl.Name, Arguments: args, ID: bl.ID, Status: "completed"})
+		}
+	}
+	if text.Len() > 0 {
+		out.Output = append([]openai.Item{{Type: "message", Role: "assistant", Content: []any{map[string]any{"type": "output_text", "text": text.String()}}}}, out.Output...)
+	}
+	// input tokens include cache writes/reads so the sum matches what is billed
+	out.Usage.InputTokens = r.Usage.InputTokens + r.Usage.CacheCreationInputTokens + r.Usage.CacheReadInputTokens
+	out.Usage.OutputTokens = r.Usage.OutputTokens
+	out.Usage.TotalTokens = out.Usage.InputTokens + out.Usage.OutputTokens
+
+	// remember the conversation for the next turn: the assistant content is replayed as the raw JSON the API returned,
+	// so tool_use ids and thinking blocks (adaptive thinking is always on for Claude 5.5; a thinking block replayed
+	// without its text/signature is rejected with "thinking.thinking: Field required") stay intact
+	full := append(append([]message{}, msgs...), message{Role: "assistant", Content: r.Content})
+	c.mu.Lock()
+	c.conv[out.ID] = conversation{system: body.System, msgs: full}
+	c.order = append(c.order, out.ID)
+	for len(c.order) > maxConvCached {
+		delete(c.conv, c.order[0])
+		c.order = c.order[1:]
+	}
+	c.mu.Unlock()
+	return out, nil
+}
+
+func truncate(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n] + "..."
+}

ai/internal/anthropic/client_test.go

@@ -0,0 +1,122 @@
+package anthropic
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
+)
+
+// Two turns through a fake Messages API: the first asks for a tool, the second (with previous_response_id and a
+// forced submit) must carry the tool_result, the assistant tool_use block, the nudge text and an auto tool_choice.
+func TestTranslateTurns(t *testing.T) {
+	var bodies []map[string]any
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/messages" || r.Header.Get("x-api-key") != "k1" || r.Header.Get("anthropic-version") != "2023-06-01" {
+			t.Errorf("bad request line/headers: %s %v", r.URL.Path, r.Header)
+		}
+		var b map[string]any
+		_ = json.NewDecoder(r.Body).Decode(&b)
+		bodies = append(bodies, b)
+		switch len(bodies) {
+		case 1:
+			w.Write([]byte(`{"id":"msg_1","type":"message","model":"claude-sonnet-5-5","stop_reason":"tool_use","content":[{"type":"thinking","thinking":"let me probe","signature":"sig-abc"},{"type":"text","text":"checking"},{"type":"tool_use","id":"toolu_1","name":"get_probe_status","input":{}}],"usage":{"input_tokens":100,"output_tokens":20,"cache_creation_input_tokens":5,"cache_read_input_tokens":0}}`))
+		default:
+			w.Write([]byte(`{"id":"msg_2","type":"message","model":"claude-sonnet-5-5","stop_reason":"tool_use","content":[{"type":"tool_use","id":"toolu_2","name":"submit_diagnosis","input":{"root_cause_label":"x","confidence":0.9}}],"usage":{"input_tokens":200,"output_tokens":30}}`))
+		}
+	}))
+	defer srv.Close()
+	c := New(srv.URL, func() (string, error) { return "k1", nil })
+	pf := false
+	tools := []openai.Tool{{Type: "function", Strict: true, Name: "get_probe_status", Description: "probe", Parameters: map[string]any{"type": "object", "properties": map[string]any{}}}}
+	r1, err := c.CreateResponse(context.Background(), openai.Request{Model: "claude-sonnet-5-5", MaxOutputTokens: 1234, ParallelToolCalls: &pf, Tools: tools, ToolChoice: "auto",
+		Input: []openai.Item{{Type: "message", Role: "system", Content: "SYS"}, {Type: "message", Role: "user", Content: "alert"}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	fcs := r1.FunctionCalls()
+	if r1.ID != "msg_1" || len(fcs) != 1 || fcs[0].CallID != "toolu_1" || fcs[0].Name != "get_probe_status" || fcs[0].Arguments != "{}" {
+		t.Fatalf("turn 1 translation: %+v", r1)
+	}
+	if r1.Usage.InputTokens != 105 || r1.Usage.OutputTokens != 20 || r1.Usage.TotalTokens != 125 {
+		t.Fatalf("usage: %+v", r1.Usage)
+	}
+	b1 := bodies[0]
+	if b1["system"] != "SYS" || b1["max_tokens"].(float64) != 1234 || b1["model"] != "claude-sonnet-5-5" {
+		t.Fatalf("body 1: %v", b1)
+	}
+	tl := b1["tools"].([]any)[0].(map[string]any)
+	if tl["name"] != "get_probe_status" || tl["input_schema"] == nil || tl["strict"] != true {
+		t.Fatalf("tool translation: %v", tl)
+	}
+	if tc := b1["tool_choice"].(map[string]any); tc["type"] != "auto" || tc["disable_parallel_tool_use"] != true {
+		t.Fatalf("tool_choice 1: %v", tc)
+	}
+
+	r2, err := c.CreateResponse(context.Background(), openai.Request{Model: "claude-sonnet-5-5", Tools: tools, PreviousResponseID: "msg_1",
+		ToolChoice: map[string]any{"type": "function", "name": "submit_diagnosis"},
+		Input:      []openai.Item{{Type: "function_call_output", CallID: "toolu_1", Output: `{"alb":1}`}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := r2.FunctionCalls(); len(got) != 1 || got[0].Name != "submit_diagnosis" || !strings.Contains(got[0].Arguments, `"root_cause_label":"x"`) {
+		t.Fatalf("turn 2 translation: %+v", r2.Output)
+	}
+	b2 := bodies[1]
+	msgs := b2["messages"].([]any)
+	if len(msgs) != 3 { // user(alert) / assistant(text+tool_use) / user(tool_result+nudge)
+		t.Fatalf("messages: %v", msgs)
+	}
+	as := msgs[1].(map[string]any)
+	if as["role"] != "assistant" || !strings.Contains(mustJSON(as["content"]), `"tool_use"`) {
+		t.Fatalf("assistant turn not replayed: %v", as)
+	}
+	// the thinking block must come back verbatim with its text and signature (Claude 5.5 rejects a stripped one)
+	if ac := mustJSON(as["content"]); !strings.Contains(ac, `"thinking":"let me probe"`) || !strings.Contains(ac, `"signature":"sig-abc"`) {
+		t.Fatalf("thinking block not replayed verbatim: %s", ac)
+	}
+	last := msgs[2].(map[string]any)["content"].([]any)
+	if tr := last[0].(map[string]any); tr["type"] != "tool_result" || tr["tool_use_id"] != "toolu_1" || tr["content"] != `{"alb":1}` {
+		t.Fatalf("tool_result: %v", tr)
+	}
+	if nd := last[1].(map[string]any); nd["type"] != "text" || !strings.Contains(nd["text"].(string), "submit_diagnosis") {
+		t.Fatalf("forced nudge missing: %v", nd)
+	}
+	if tc := b2["tool_choice"].(map[string]any); tc["type"] != "auto" {
+		t.Fatalf("forced tool_choice must become auto for Claude 5.5: %v", tc)
+	}
+	if b2["system"] != "SYS" {
+		t.Fatalf("system prompt must be carried on the chained turn, got %v", b2["system"])
+	}
+}
+
+func TestEmptyKeyAndErrors(t *testing.T) {
+	c := New("http://127.0.0.1:1", func() (string, error) { return "", nil })
+	if c.KeyPresent() {
+		t.Fatal("empty key reported present")
+	}
+	_, err := c.CreateResponse(context.Background(), openai.Request{Model: "claude-haiku-5-5", Input: []openai.Item{{Type: "message", Role: "user", Content: "x"}}})
+	if err == nil || !strings.Contains(err.Error(), "anthropic api key is empty") {
+		t.Fatalf("expected empty-key error, got %v", err)
+	}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(400)
+		w.Write([]byte(`{"type":"error","error":{"type":"invalid_request_error","message":"tool_choice: type \"tool\" and \"any\" are not supported for this model."}}`))
+	}))
+	defer srv.Close()
+	c2 := New(srv.URL, func() (string, error) { return "k", nil })
+	_, err = c2.CreateResponse(context.Background(), openai.Request{Model: "claude-haiku-5-5", Input: []openai.Item{{Type: "message", Role: "user", Content: "x"}}})
+	if err == nil || !strings.Contains(err.Error(), "status 400") || !strings.Contains(err.Error(), "invalid_request_error") {
+		t.Fatalf("expected 400 error, got %v", err)
+	}
+	_, err = c2.CreateResponse(context.Background(), openai.Request{Model: "claude-haiku-5-5", PreviousResponseID: "nope", Input: []openai.Item{{Type: "function_call_output", CallID: "a", Output: "b"}}})
+	if err == nil || !strings.Contains(err.Error(), "unknown previous_response_id") {
+		t.Fatalf("expected unknown id error, got %v", err)
+	}
+}
+
+func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) }

ai/internal/anthropic/federation.go

@@ -0,0 +1,177 @@
+package anthropic
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Federation authenticates with Workload Identity Federation instead of an API key (platform.claude.com/docs/en/
+// manage-claude/wif-providers/gcp): the VM's Google-signed identity token (metadata server, audience
+// https://api.anthropic.com, format=full so the email claim is present) is exchanged at POST /v1/oauth/token
+// (RFC 7523 jwt-bearer grant) for a short-lived sk-ant-oat01-... access token bound to an Anthropic service account.
+// The four ids are not secrets; they come from ai.env. The token is cached and re-exchanged 120 s before expiry
+// (the SDKs' advisory refresh), and a 401 from the API triggers one re-exchange. A fresh identity token is fetched
+// for every exchange (identity tokens with a jti claim are single-use on the Anthropic side).
+type Federation struct {
+	RuleID           string // fdrl_...
+	OrgID            string // Anthropic organization id (uuid)
+	ServiceAccountID string // svac_...
+	WorkspaceID      string // wrkspc_... (optional when the rule covers one workspace)
+	IdentityTokenURL string // metadata server URL (tests override it)
+	HTTP             *http.Client
+
+	mu     sync.Mutex
+	token  string
+	expiry time.Time
+	nex    int // exchanges performed (tests)
+}
+
+const defaultIdentityTokenURL = "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?audience=https://api.anthropic.com&format=full"
+
+const refreshBefore = 120 * time.Second
+
+// Configured reports whether the mandatory ids are set (then federation takes precedence over the key file).
+func (f *Federation) Configured() bool {
+	return f != nil && f.RuleID != "" && f.OrgID != "" && f.ServiceAccountID != ""
+}
+
+func (f *Federation) identityToken(ctx context.Context) (string, error) {
+	u := f.IdentityTokenURL
+	if u == "" {
+		u = defaultIdentityTokenURL
+	}
+	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Metadata-Flavor", "Google")
+	resp, err := f.http().Do(req)
+	if err != nil {
+		return "", fmt.Errorf("identity token: %w", err)
+	}
+	defer resp.Body.Close()
+	b, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+	tok := strings.TrimSpace(string(b))
+	if resp.StatusCode != 200 || strings.Count(tok, ".") != 2 {
+		return "", fmt.Errorf("identity token: status %d: %s", resp.StatusCode, truncate(tok, 200))
+	}
+	return tok, nil
+}
+
+func (f *Federation) http() *http.Client {
+	if f.HTTP != nil {
+		return f.HTTP
+	}
+	return &http.Client{Timeout: 30 * time.Second}
+}
+
+// AccessToken returns a cached access token, exchanging a fresh identity token when none is cached or the cached
+// one is within refreshBefore of its expiry.
+func (f *Federation) AccessToken(ctx context.Context, baseURL string) (string, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.token != "" && time.Until(f.expiry) > refreshBefore {
+		return f.token, nil
+	}
+	tok, exp, err := f.exchange(ctx, baseURL)
+	if err != nil {
+		return "", err
+	}
+	f.token, f.expiry = tok, exp
+	f.nex++
+	return tok, nil
+}
+
+// Invalidate drops the cached token (after a 401) so the next call exchanges again.
+func (f *Federation) Invalidate() {
+	f.mu.Lock()
+	f.token, f.expiry = "", time.Time{}
+	f.mu.Unlock()
+}
+
+// Exchanges counts the token exchanges performed (tests).
+func (f *Federation) Exchanges() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.nex
+}
+
+type tokenResponse struct {
+	AccessToken string `json:"access_token"`
+	TokenType   string `json:"token_type"`
+	ExpiresIn   int    `json:"expires_in"`
+	Error       string `json:"error"`
+	ErrorDesc   string `json:"error_description"`
+	ErrorObj    *struct {
+		Type    string `json:"type"`
+		Message string `json:"message"`
+	} `json:"error_obj,omitempty"`
+}
+
+func (f *Federation) exchange(ctx context.Context, baseURL string) (string, time.Time, error) {
+	jwt, err := f.identityToken(ctx)
+	if err != nil {
+		return "", time.Time{}, err
+	}
+	body := map[string]string{
+		"grant_type":         "urn:ietf:params:oauth:grant-type:jwt-bearer",
+		"assertion":          jwt,
+		"federation_rule_id": f.RuleID,
+		"organization_id":    f.OrgID,
+		"service_account_id": f.ServiceAccountID,
+	}
+	if f.WorkspaceID != "" {
+		body["workspace_id"] = f.WorkspaceID
+	}
+	b, _ := json.Marshal(body)
+	req, err := http.NewRequestWithContext(ctx, "POST", strings.TrimRight(baseURL, "/")+"/v1/oauth/token", bytes.NewReader(b))
+	if err != nil {
+		return "", time.Time{}, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := f.http().Do(req)
+	if err != nil {
+		return "", time.Time{}, fmt.Errorf("token exchange: %w", err)
+	}
+	defer resp.Body.Close()
+	rb, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	var tr tokenResponse
+	_ = json.Unmarshal(rb, &tr)
+	if resp.StatusCode/100 != 2 || tr.AccessToken == "" {
+		// the API answers a failed exchange with an opaque 401 "Authentication failed"; the deny reason is on the
+		// Console authentication-history page (platform.claude.com/settings/workload-identity-federation?tab=history)
+		return "", time.Time{}, fmt.Errorf("token exchange: status %d: %s", resp.StatusCode, truncate(strings.TrimSpace(string(rb)), 300))
+	}
+	ttl := time.Duration(tr.ExpiresIn) * time.Second
+	if ttl <= 0 {
+		ttl = 60 * time.Second
+	}
+	return tr.AccessToken, time.Now().Add(ttl), nil
+}
+
+// Probe performs one exchange and reports the outcome (selfcheck): the setup can be verified without a diagnosis.
+func (f *Federation) Probe(ctx context.Context, baseURL string) map[string]any {
+	out := map[string]any{"rule_id": f.RuleID, "service_account_id": f.ServiceAccountID}
+	f.Invalidate()
+	start := time.Now()
+	tok, err := f.AccessToken(ctx, baseURL)
+	out["ms"] = time.Since(start).Milliseconds()
+	if err != nil {
+		out["ok"] = false
+		out["error"] = err.Error()
+		return out
+	}
+	f.mu.Lock()
+	out["ok"] = true
+	out["token_prefix"] = truncate(tok, 12)
+	out["expires_in_s"] = int(time.Until(f.expiry).Seconds())
+	f.mu.Unlock()
+	return out
+}

ai/internal/anthropic/federation_test.go

@@ -0,0 +1,130 @@
+package anthropic
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
+)
+
+// Federation: the identity token from the (fake) metadata server is exchanged once at /v1/oauth/token with the
+// jwt-bearer grant and the four ids; /v1/messages gets "Authorization: Bearer" and no x-api-key; the cached token is
+// reused; a 401 triggers exactly one re-exchange; an expired cache re-exchanges before the next call.
+func TestFederationExchangeAndReuse(t *testing.T) {
+	var idCalls, exchanges, msgCalls int32
+	var fail401 atomic.Bool
+	md := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Metadata-Flavor") != "Google" || !strings.Contains(r.URL.RawQuery, "audience=https://api.anthropic.com") || !strings.Contains(r.URL.RawQuery, "format=full") {
+			t.Errorf("bad metadata request: %s %v", r.URL.String(), r.Header)
+		}
+		atomic.AddInt32(&idCalls, 1)
+		w.Write([]byte("eyJhbGciOiJSUzI1NiJ9.eyJpc3MiOiJodHRwczovL2FjY291bnRzLmdvb2dsZS5jb20ifQ.sig"))
+	}))
+	defer md.Close()
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/v1/oauth/token":
+			var b map[string]string
+			_ = json.NewDecoder(r.Body).Decode(&b)
+			if b["grant_type"] != "urn:ietf:params:oauth:grant-type:jwt-bearer" || !strings.HasPrefix(b["assertion"], "eyJ") || b["federation_rule_id"] != "fdrl_1" || b["organization_id"] != "org-1" || b["service_account_id"] != "svac_1" || b["workspace_id"] != "wrkspc_1" {
+				t.Errorf("bad exchange body: %v", b)
+			}
+			n := atomic.AddInt32(&exchanges, 1)
+			w.Write([]byte(`{"access_token":"sk-ant-oat01-tok` + string(rune('0'+n)) + `","token_type":"Bearer","expires_in":600}`))
+		case "/v1/messages":
+			atomic.AddInt32(&msgCalls, 1)
+			if r.Header.Get("x-api-key") != "" || !strings.HasPrefix(r.Header.Get("Authorization"), "Bearer sk-ant-oat01-tok") {
+				t.Errorf("bad auth headers: %v", r.Header)
+			}
+			if fail401.Swap(false) {
+				w.WriteHeader(401)
+				w.Write([]byte(`{"type":"error","error":{"type":"authentication_error","message":"Authentication failed"}}`))
+				return
+			}
+			w.Write([]byte(`{"id":"msg_f","type":"message","model":"claude-sonnet-5-5","stop_reason":"end_turn","content":[{"type":"text","text":"ok"}],"usage":{"input_tokens":3,"output_tokens":1}}`))
+		default:
+			t.Errorf("unexpected path %s", r.URL.Path)
+		}
+	}))
+	defer api.Close()
+	c := New(api.URL, func() (string, error) { return "", nil }) // no key file: federation must carry the call
+	fed := &Federation{RuleID: "fdrl_1", OrgID: "org-1", ServiceAccountID: "svac_1", WorkspaceID: "wrkspc_1", IdentityTokenURL: md.URL + "/identity?audience=https://api.anthropic.com&format=full"}
+	c.SetFederation(fed)
+	if c.AuthMode() != "federation" {
+		t.Fatalf("auth mode %s", c.AuthMode())
+	}
+	req := openai.Request{Model: "claude-sonnet-5-5", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}}
+	for i := 0; i < 2; i++ {
+		if _, err := c.CreateResponse(context.Background(), req); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if atomic.LoadInt32(&exchanges) != 1 || atomic.LoadInt32(&idCalls) != 1 || atomic.LoadInt32(&msgCalls) != 2 {
+		t.Fatalf("expected one exchange for two calls, got exchanges=%d id=%d msgs=%d", exchanges, idCalls, msgCalls)
+	}
+	// 401 -> one re-exchange and a successful retry
+	fail401.Store(true)
+	if _, err := c.CreateResponse(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+	if atomic.LoadInt32(&exchanges) != 2 || atomic.LoadInt32(&msgCalls) != 4 {
+		t.Fatalf("401 must re-exchange once: exchanges=%d msgs=%d", exchanges, msgCalls)
+	}
+	// cached token close to expiry -> advisory refresh before the next call
+	fed.mu.Lock()
+	fed.expiry = time.Now().Add(60 * time.Second)
+	fed.mu.Unlock()
+	if _, err := c.CreateResponse(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+	if atomic.LoadInt32(&exchanges) != 3 {
+		t.Fatalf("near-expiry token must be refreshed: exchanges=%d", exchanges)
+	}
+	// probe reports a successful exchange
+	p := fed.Probe(context.Background(), api.URL)
+	if p["ok"] != true || p["expires_in_s"].(int) < 500 || !strings.HasPrefix(p["token_prefix"].(string), "sk-ant-oat01") {
+		t.Fatalf("probe: %v", p)
+	}
+}
+
+func TestFederationFailuresAndPrecedence(t *testing.T) {
+	// incomplete ids are ignored: the key path stays active
+	c := New("http://127.0.0.1:1", func() (string, error) { return "k", nil })
+	c.SetFederation(&Federation{RuleID: "fdrl_1"})
+	if c.AuthMode() != "api_key" || c.Federation() != nil {
+		t.Fatalf("incomplete federation must not activate: %s", c.AuthMode())
+	}
+	// exchange denied (opaque 401) -> clear error, no message call
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/oauth/token" {
+			w.WriteHeader(401)
+			w.Write([]byte(`{"type":"error","error":{"type":"authentication_error","message":"Authentication failed"}}`))
+			return
+		}
+		t.Errorf("messages must not be called when the exchange fails")
+	}))
+	defer api.Close()
+	md := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("a.b.c")) }))
+	defer md.Close()
+	c2 := New(api.URL, func() (string, error) { return "", nil })
+	fed := &Federation{RuleID: "fdrl_1", OrgID: "org-1", ServiceAccountID: "svac_1", IdentityTokenURL: md.URL}
+	c2.SetFederation(fed)
+	_, err := c2.CreateResponse(context.Background(), openai.Request{Model: "claude-haiku-5-5", Input: []openai.Item{{Type: "message", Role: "user", Content: "x"}}})
+	if err == nil || !strings.Contains(err.Error(), "token exchange: status 401") {
+		t.Fatalf("expected exchange failure, got %v", err)
+	}
+	if p := fed.Probe(context.Background(), api.URL); p["ok"] != false || !strings.Contains(p["error"].(string), "status 401") {
+		t.Fatalf("probe must report the denial: %v", p)
+	}
+	// metadata server unreachable
+	fed2 := &Federation{RuleID: "fdrl_1", OrgID: "org-1", ServiceAccountID: "svac_1", IdentityTokenURL: "http://127.0.0.1:1/identity", HTTP: &http.Client{Timeout: 2 * time.Second}}
+	if _, err := fed2.AccessToken(context.Background(), api.URL); err == nil || !strings.Contains(err.Error(), "identity token") {
+		t.Fatalf("expected identity token error, got %v", err)
+	}
+}

ai/internal/external/responder.go

@@ -0,0 +1,359 @@
+// Package external lets a client outside the AI VM act as the LLM for model ids starting with "external-": the
+// diagnosis loop (internal/agent) is unchanged, but every turn is queued as a Turn and served to the client over
+// GET /external/next (long poll); the client answers with POST /external/reply. Used for reference measurements with
+// a model that has no API path (2026-10-16: Claude Mythos 5.1 driven by a Claude Code subagent). Token usage is not
+// known for such turns and is reported as zero; reports mark those runs as reference records.
+//
+// A reply becomes a diagnosis and, with auto-approval, an executed action, so whoever can answer a turn controls the
+// lab: the endpoints are served on their own listener (EXTERNAL_ADDR, off by default; the VM maps it to 127.0.0.1 so
+// only the IAP SSH tunnel reaches it), turn ids are random, and every reply must carry the per-turn reply_token that
+// only the client who took the turn received.
+package external
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/subtle"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
+)
+
+const maxConvCached = 64
+
+// Turn is one request to the external client: the whole conversation so far (the loop sends the system prompt only
+// on the first turn, so it is carried along), the tool definitions and whether the final submit_diagnosis is forced.
+type Turn struct {
+	TurnID          string        `json:"turn_id"`
+	ReplyToken      string        `json:"reply_token"` // random, handed out with the turn, required by /external/reply
+	Requeued        int           `json:"requeued"`    // times this turn was handed out again after an unanswered take
+	Session         string        `json:"session"`     // first response id of the conversation ("" on the first turn)
+	Model           string        `json:"model"`
+	Turn            int           `json:"turn"` // 1-based index within the conversation
+	System          string        `json:"system"`
+	Input           []openai.Item `json:"input"` // user messages, assistant function_call items, function_call_output items
+	Tools           []openai.Tool `json:"tools"`
+	ForceSubmit     bool          `json:"force_submit"` // the loop asked for submit_diagnosis now (last turn)
+	MaxOutputTokens int           `json:"max_output_tokens"`
+	IssuedAt        time.Time     `json:"issued_at"`
+	DeadlineAt      time.Time     `json:"deadline_at"`
+}
+
+// Reply is the client's answer: function calls (name + arguments) and/or assistant text.
+type Reply struct {
+	TurnID     string      `json:"turn_id"`
+	ReplyToken string      `json:"reply_token"`
+	Output     []replyItem `json:"output"`
+	Error      string      `json:"error,omitempty"` // client gave up on this turn
+}
+
+type replyItem struct {
+	Type      string          `json:"type"` // function_call | message
+	Name      string          `json:"name,omitempty"`
+	Arguments json.RawMessage `json:"arguments,omitempty"` // JSON object or a JSON string holding one
+	Text      string          `json:"text,omitempty"`
+}
+
+type pending struct {
+	turn    Turn
+	reply   chan Reply
+	takenAt time.Time // zero while queued
+}
+
+type conversation struct {
+	system string
+	items  []openai.Item
+	turn   int
+	first  string
+}
+
+// Responder implements agent.Responder; one instance serves every external-* diagnosis.
+type Responder struct {
+	TurnTimeout  time.Duration // how long one turn waits for the client (bounded by the caller's context as well)
+	RequeueAfter time.Duration // a taken turn with no reply after this long goes back to the queue with a new token
+
+	mu       sync.Mutex
+	queue    []*pending          // issued, not yet taken by a client
+	taken    map[string]*pending // taken, waiting for the reply
+	conv     map[string]conversation
+	order    []string
+	seq      int
+	notify   chan struct{}
+	lastPoll time.Time
+	lastTurn time.Time
+}
+
+func New(turnTimeout time.Duration) *Responder {
+	if turnTimeout <= 0 {
+		turnTimeout = 20 * time.Minute
+	}
+	return &Responder{TurnTimeout: turnTimeout, RequeueAfter: 3 * time.Minute, taken: map[string]*pending{}, conv: map[string]conversation{}, notify: make(chan struct{}, 1)}
+}
+
+// CreateResponse queues the turn and blocks until the client replies, the context ends or TurnTimeout passes.
+func (r *Responder) CreateResponse(ctx context.Context, req openai.Request) (*openai.Response, error) {
+	var hist conversation
+	if req.PreviousResponseID != "" {
+		r.mu.Lock()
+		prev, ok := r.conv[req.PreviousResponseID]
+		r.mu.Unlock()
+		if !ok {
+			return nil, fmt.Errorf("external: unknown previous_response_id %q", req.PreviousResponseID)
+		}
+		hist = conversation{system: prev.system, items: append([]openai.Item{}, prev.items...), turn: prev.turn, first: prev.first}
+	}
+	for _, it := range req.Input {
+		if it.Type == "message" && it.Role == "system" {
+			if s, _ := it.Content.(string); s != "" {
+				if hist.system != "" {
+					hist.system += "\n\n"
+				}
+				hist.system += s
+			}
+			continue
+		}
+		hist.items = append(hist.items, it)
+	}
+	if len(hist.items) == 0 {
+		return nil, fmt.Errorf("external: no input items")
+	}
+	forced := false
+	if m, ok := req.ToolChoice.(map[string]any); ok && m["type"] == "function" {
+		forced = true
+	}
+	id, tok := "ext_"+randHex(8), randHex(16)
+	r.mu.Lock()
+	r.seq++
+	now := time.Now()
+	p := &pending{turn: Turn{TurnID: id, ReplyToken: tok, Session: hist.first, Model: req.Model, Turn: hist.turn + 1, System: hist.system, Input: hist.items, Tools: req.Tools,
+		ForceSubmit: forced, MaxOutputTokens: req.MaxOutputTokens, IssuedAt: now, DeadlineAt: now.Add(r.TurnTimeout)}, reply: make(chan Reply, 1)}
+	r.queue = append(r.queue, p)
+	r.lastTurn = now
+	r.mu.Unlock()
+	select {
+	case r.notify <- struct{}{}:
+	default:
+	}
+
+	var rep Reply
+	select {
+	case rep = <-p.reply:
+	case <-ctx.Done():
+		r.drop(id)
+		return nil, fmt.Errorf("external: turn %s not answered before the diagnosis deadline (%v)", id, ctx.Err())
+	case <-time.After(r.TurnTimeout):
+		r.drop(id)
+		return nil, fmt.Errorf("external: turn %s not answered within %s", id, r.TurnTimeout)
+	}
+	if rep.Error != "" {
+		return nil, fmt.Errorf("external: client error on turn %s: %s", id, rep.Error)
+	}
+	out := &openai.Response{ID: id, Status: "completed", Model: req.Model}
+	var text strings.Builder
+	for i, it := range rep.Output {
+		switch it.Type {
+		case "function_call":
+			out.Output = append(out.Output, openai.Item{Type: "function_call", CallID: fmt.Sprintf("%s_call_%d", id, i+1), ID: fmt.Sprintf("%s_call_%d", id, i+1), Name: it.Name, Arguments: argString(it.Arguments), Status: "completed"})
+		case "message":
+			text.WriteString(it.Text)
+		}
+	}
+	if text.Len() > 0 {
+		out.Output = append([]openai.Item{{Type: "message", Role: "assistant", Content: []any{map[string]any{"type": "output_text", "text": text.String()}}}}, out.Output...)
+	}
+	// remember the conversation (assistant output items are replayed to the client on the next turn)
+	first := hist.first
+	if first == "" {
+		first = id
+	}
+	items := append(append([]openai.Item{}, hist.items...), out.Output...)
+	r.mu.Lock()
+	r.conv[id] = conversation{system: hist.system, items: items, turn: hist.turn + 1, first: first}
+	r.order = append(r.order, id)
+	for len(r.order) > maxConvCached {
+		delete(r.conv, r.order[0])
+		r.order = r.order[1:]
+	}
+	r.mu.Unlock()
+	return out, nil
+}
+
+// argString returns the function arguments as the JSON object text (a JSON string value is unwrapped).
+func argString(raw json.RawMessage) string {
+	s := strings.TrimSpace(string(raw))
+	if s == "" {
+		return "{}"
+	}
+	if strings.HasPrefix(s, `"`) {
+		var inner string
+		if json.Unmarshal(raw, &inner) == nil {
+			if strings.TrimSpace(inner) == "" {
+				return "{}"
+			}
+			return inner
+		}
+	}
+	return s
+}
+
+// requeueStale (mu held) returns taken turns whose reply never came (client lost the output, crashed, was replaced)
+// to the front of the queue with a fresh reply token, so the next poll serves them again; the old token is void.
+func (r *Responder) requeueStale() {
+	if r.RequeueAfter <= 0 {
+		return
+	}
+	var stale []*pending
+	for id, p := range r.taken {
+		if time.Since(p.takenAt) >= r.RequeueAfter {
+			delete(r.taken, id)
+			p.turn.ReplyToken = randHex(16)
+			p.turn.Requeued++
+			p.takenAt = time.Time{}
+			stale = append(stale, p)
+		}
+	}
+	if len(stale) > 0 {
+		r.queue = append(stale, r.queue...)
+	}
+}
+
+func (r *Responder) drop(id string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	delete(r.taken, id)
+	for i, p := range r.queue {
+		if p.turn.TurnID == id {
+			r.queue = append(r.queue[:i], r.queue[i+1:]...)
+			return
+		}
+	}
+}
+
+// Next is GET /external/next?wait=<s>: the oldest queued turn (200), or 204 when none arrived within wait seconds.
+func (r *Responder) Next(w http.ResponseWriter, req *http.Request) {
+	wait := 20 * time.Second
+	if v, err := strconv.Atoi(req.URL.Query().Get("wait")); err == nil && v >= 0 && v <= 60 {
+		wait = time.Duration(v) * time.Second
+	}
+	deadline := time.NewTimer(wait)
+	defer deadline.Stop()
+	for {
+		r.mu.Lock()
+		r.lastPoll = time.Now()
+		r.requeueStale()
+		if len(r.queue) > 0 {
+			p := r.queue[0]
+			r.queue = r.queue[1:]
+			p.takenAt = time.Now()
+			r.taken[p.turn.TurnID] = p
+			r.mu.Unlock()
+			writeJSON(w, 200, p.turn)
+			return
+		}
+		r.mu.Unlock()
+		select {
+		case <-r.notify:
+		case <-deadline.C:
+			w.WriteHeader(204)
+			return
+		case <-req.Context().Done():
+			return
+		}
+	}
+}
+
+// ReplyHandler is POST /external/reply with a Reply body; 404 when the turn is unknown (already answered, dropped
+// after its deadline, or never taken through /external/next).
+func (r *Responder) ReplyHandler(w http.ResponseWriter, req *http.Request) {
+	var rep Reply
+	if err := json.NewDecoder(http.MaxBytesReader(w, req.Body, 1<<20)).Decode(&rep); err != nil {
+		writeJSON(w, 400, map[string]string{"error": err.Error()})
+		return
+	}
+	if rep.Error == "" {
+		n := 0
+		for _, it := range rep.Output {
+			switch it.Type {
+			case "function_call":
+				if it.Name == "" {
+					writeJSON(w, 400, map[string]string{"error": "function_call without name"})
+					return
+				}
+				if a := argString(it.Arguments); !json.Valid([]byte(a)) || !strings.HasPrefix(strings.TrimSpace(a), "{") {
+					writeJSON(w, 400, map[string]string{"error": "function_call arguments must be a JSON object"})
+					return
+				}
+				n++
+			case "message":
+				n++
+			default:
+				writeJSON(w, 400, map[string]string{"error": "output item type must be function_call or message"})
+				return
+			}
+		}
+		if n == 0 {
+			writeJSON(w, 400, map[string]string{"error": "output must hold at least one function_call or message"})
+			return
+		}
+	}
+	r.mu.Lock()
+	p, ok := r.taken[rep.TurnID]
+	if ok && subtle.ConstantTimeCompare([]byte(p.turn.ReplyToken), []byte(rep.ReplyToken)) != 1 {
+		r.mu.Unlock()
+		writeJSON(w, 403, map[string]string{"error": "reply_token does not match turn " + rep.TurnID})
+		return
+	}
+	if ok {
+		delete(r.taken, rep.TurnID)
+	}
+	r.mu.Unlock()
+	if !ok {
+		writeJSON(w, 404, map[string]string{"error": "unknown or expired turn_id " + rep.TurnID})
+		return
+	}
+	p.reply <- rep
+	writeJSON(w, 200, map[string]any{"ok": true, "turn_id": rep.TurnID})
+}
+
+// Status is shown by /selfcheck.
+func (r *Responder) Status() map[string]any {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := map[string]any{"queued": len(r.queue), "awaiting_reply": len(r.taken), "turn_timeout_s": int(r.TurnTimeout.Seconds()), "requeue_after_s": int(r.RequeueAfter.Seconds())}
+	if !r.lastPoll.IsZero() {
+		out["last_poll_age_s"] = int(time.Since(r.lastPoll).Seconds())
+	}
+	if !r.lastTurn.IsZero() {
+		out["last_turn_age_s"] = int(time.Since(r.lastTurn).Seconds())
+	}
+	return out
+}
+
+func randHex(n int) string {
+	b := make([]byte, n)
+	if _, err := rand.Read(b); err != nil {
+		panic("crypto/rand: " + err.Error())
+	}
+	return hex.EncodeToString(b)
+}
+
+// Handler serves the two endpoints; mount it on the dedicated external listener only.
+func (r *Responder) Handler() http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /external/next", r.Next)
+	mux.HandleFunc("POST /external/reply", r.ReplyHandler)
+	return mux
+}
+
+func writeJSON(w http.ResponseWriter, code int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(code)
+	_ = json.NewEncoder(w).Encode(v)
+}

ai/internal/external/responder_test.go

@@ -0,0 +1,222 @@
+package external
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai"
+)
+
+// Two loop turns through the HTTP surface: the client takes turn 1 (system + user), answers with a tool call,
+// then turn 2 (with previous_response_id) must replay the assistant call and the function_call_output and carry
+// the forced-submit flag; the reply's arguments may be a JSON string or object.
+func TestTurnsOverHTTP(t *testing.T) {
+	r := New(5 * time.Second)
+	srv := httptest.NewServer(r.Handler())
+	defer srv.Close()
+	tools := []openai.Tool{{Type: "function", Strict: true, Name: "get_probe_status", Parameters: map[string]any{"type": "object"}}}
+	type res struct {
+		resp *openai.Response
+		err  error
+	}
+	ch := make(chan res, 1)
+	go func() {
+		resp, err := r.CreateResponse(context.Background(), openai.Request{Model: "external-claude-mythos-5-1", Tools: tools, ToolChoice: "auto", MaxOutputTokens: 4000,
+			Input: []openai.Item{{Type: "message", Role: "system", Content: "SYS"}, {Type: "message", Role: "user", Content: "alert"}}})
+		ch <- res{resp, err}
+	}()
+	turn := poll(t, srv.URL)
+	if !strings.HasPrefix(turn.TurnID, "ext_") || len(turn.TurnID) != 20 || len(turn.ReplyToken) != 32 || turn.Session != "" || turn.Turn != 1 || turn.System != "SYS" || len(turn.Input) != 1 || turn.Input[0].Content != "alert" || len(turn.Tools) != 1 || turn.ForceSubmit || turn.MaxOutputTokens != 4000 {
+		t.Fatalf("turn 1: %+v", turn)
+	}
+	// second poll while nothing is queued -> 204 within the wait
+	if code := get(t, srv.URL+"/external/next?wait=0"); code != 204 {
+		t.Fatalf("empty queue must answer 204, got %d", code)
+	}
+	// a reply without the turn's token is refused (the turn stays open)
+	if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn.TurnID+`","reply_token":"wrong","output":[{"type":"message","text":"x"}]}`); code != 403 {
+		t.Fatalf("wrong token must be 403, got %d", code)
+	}
+	code, body := post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn.TurnID+`","reply_token":"`+turn.ReplyToken+`","output":[{"type":"message","text":"checking"},{"type":"function_call","name":"get_probe_status","arguments":"{}"}]}`)
+	if code != 200 {
+		t.Fatalf("reply 1: %d %s", code, body)
+	}
+	r1 := <-ch
+	if r1.err != nil {
+		t.Fatal(r1.err)
+	}
+	fcs := r1.resp.FunctionCalls()
+	if r1.resp.ID != turn.TurnID || len(fcs) != 1 || fcs[0].Name != "get_probe_status" || fcs[0].Arguments != "{}" || fcs[0].CallID == "" || r1.resp.Usage.TotalTokens != 0 {
+		t.Fatalf("response 1: %+v", r1.resp)
+	}
+	if !strings.Contains(r1.resp.Text(), "checking") {
+		t.Fatalf("text not carried: %q", r1.resp.Text())
+	}
+
+	go func() {
+		resp, err := r.CreateResponse(context.Background(), openai.Request{Model: "external-claude-mythos-5-1", Tools: tools, PreviousResponseID: turn.TurnID,
+			ToolChoice: map[string]any{"type": "function", "name": "submit_diagnosis"},
+			Input:      []openai.Item{{Type: "function_call_output", CallID: fcs[0].CallID, Output: `{"alb":1}`}}})
+		ch <- res{resp, err}
+	}()
+	turn2 := poll(t, srv.URL)
+	if turn2.TurnID == turn.TurnID || turn2.Session != turn.TurnID || turn2.Turn != 2 || turn2.System != "SYS" || !turn2.ForceSubmit {
+		t.Fatalf("turn 2 header: %+v", turn2)
+	}
+	// history: user alert / assistant text / assistant function_call / function_call_output
+	if len(turn2.Input) != 4 || turn2.Input[2].Type != "function_call" || turn2.Input[2].CallID != fcs[0].CallID || turn2.Input[3].Type != "function_call_output" || turn2.Input[3].Output != `{"alb":1}` {
+		t.Fatalf("turn 2 history: %s", mustJSON(turn2.Input))
+	}
+	code, body = post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn2.TurnID+`","reply_token":"`+turn2.ReplyToken+`","output":[{"type":"function_call","name":"submit_diagnosis","arguments":{"root_cause_label":"x","confidence":0.9}}]}`)
+	if code != 200 {
+		t.Fatalf("reply 2: %d %s", code, body)
+	}
+	r2 := <-ch
+	if r2.err != nil {
+		t.Fatal(r2.err)
+	}
+	if got := r2.resp.FunctionCalls(); len(got) != 1 || got[0].Name != "submit_diagnosis" || !strings.Contains(got[0].Arguments, `"root_cause_label":"x"`) {
+		t.Fatalf("response 2: %+v", r2.resp.Output)
+	}
+	// an answered turn cannot be answered twice; malformed replies are refused before touching the queue
+	if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn2.TurnID+`","reply_token":"`+turn2.ReplyToken+`","output":[{"type":"message","text":"again"}]}`); code != 404 {
+		t.Fatalf("second reply must be 404, got %d", code)
+	}
+	if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"ext_9","output":[{"type":"function_call","name":"x","arguments":"[1]"}]}`); code != 400 {
+		t.Fatalf("non-object arguments must be 400, got %d", code)
+	}
+	if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"ext_9","output":[]}`); code != 400 {
+		t.Fatalf("empty output must be 400, got %d", code)
+	}
+	st := r.Status()
+	if st["queued"] != 0 || st["awaiting_reply"] != 0 {
+		t.Fatalf("status: %v", st)
+	}
+}
+
+// Deadlines: the diagnosis context ending drops the queued turn with an error; a client "error" reply surfaces as an
+// error; an unknown previous_response_id is refused.
+func TestDeadlinesAndErrors(t *testing.T) {
+	r := New(300 * time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	_, err := r.CreateResponse(ctx, openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}})
+	if err == nil || !strings.Contains(err.Error(), "not answered before the diagnosis deadline") {
+		t.Fatalf("expected deadline error, got %v", err)
+	}
+	if st := r.Status(); st["queued"] != 0 {
+		t.Fatalf("dropped turn still queued: %v", st)
+	}
+	_, err = r.CreateResponse(context.Background(), openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}})
+	if err == nil || !strings.Contains(err.Error(), "not answered within") {
+		t.Fatalf("expected turn timeout, got %v", err)
+	}
+	// error reply
+	r2 := New(5 * time.Second)
+	srv := httptest.NewServer(r2.Handler())
+	defer srv.Close()
+	ch := make(chan error, 1)
+	go func() {
+		_, err := r2.CreateResponse(context.Background(), openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}})
+		ch <- err
+	}()
+	turn := poll(t, srv.URL)
+	post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn.TurnID+`","reply_token":"`+turn.ReplyToken+`","error":"gave up"}`)
+	if err := <-ch; err == nil || !strings.Contains(err.Error(), "gave up") {
+		t.Fatalf("expected client error, got %v", err)
+	}
+	_, err = r2.CreateResponse(context.Background(), openai.Request{Model: "external-x", PreviousResponseID: "nope", Input: []openai.Item{{Type: "function_call_output", CallID: "a", Output: "b"}}})
+	if err == nil || !strings.Contains(err.Error(), "unknown previous_response_id") {
+		t.Fatalf("expected unknown id error, got %v", err)
+	}
+}
+
+// A taken turn whose reply never comes is handed out again after RequeueAfter with a new token; the old token is
+// refused and the new one completes the turn.
+func TestRequeueUnansweredTurn(t *testing.T) {
+	r := New(10 * time.Second)
+	r.RequeueAfter = 200 * time.Millisecond
+	srv := httptest.NewServer(r.Handler())
+	defer srv.Close()
+	ch := make(chan error, 1)
+	go func() {
+		resp, err := r.CreateResponse(context.Background(), openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}})
+		if err == nil && len(resp.FunctionCalls()) != 1 {
+			err = fmt.Errorf("unexpected response %+v", resp)
+		}
+		ch <- err
+	}()
+	first := poll(t, srv.URL) // taken, never answered
+	if first.Requeued != 0 {
+		t.Fatalf("fresh turn marked requeued: %+v", first)
+	}
+	if code := get(t, srv.URL+"/external/next?wait=0"); code != 204 {
+		t.Fatalf("taken turn must not be served again at once, got %d", code)
+	}
+	time.Sleep(250 * time.Millisecond)
+	second := poll(t, srv.URL)
+	if second.TurnID != first.TurnID || second.ReplyToken == first.ReplyToken || second.Requeued != 1 {
+		t.Fatalf("requeue: first=%+v second=%+v", first, second)
+	}
+	if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+first.TurnID+`","reply_token":"`+first.ReplyToken+`","output":[{"type":"function_call","name":"x","arguments":{}}]}`); code != 403 {
+		t.Fatalf("old token must be refused, got %d", code)
+	}
+	if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+second.TurnID+`","reply_token":"`+second.ReplyToken+`","output":[{"type":"function_call","name":"x","arguments":{}}]}`); code != 200 {
+		t.Fatalf("new token must be accepted, got %d", code)
+	}
+	if err := <-ch; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func poll(t *testing.T, base string) Turn {
+	t.Helper()
+	for i := 0; i < 20; i++ {
+		resp, err := http.Get(base + "/external/next?wait=1")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if resp.StatusCode == 200 {
+			var turn Turn
+			if err := json.NewDecoder(resp.Body).Decode(&turn); err != nil {
+				t.Fatal(err)
+			}
+			resp.Body.Close()
+			return turn
+		}
+		resp.Body.Close()
+	}
+	t.Fatal("no turn served")
+	return Turn{}
+}
+
+func get(t *testing.T, u string) int {
+	t.Helper()
+	resp, err := http.Get(u)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+	return resp.StatusCode
+}
+
+func post(t *testing.T, u, body string) (int, string) {
+	t.Helper()
+	resp, err := http.Post(u, "application/json", bytes.NewReader([]byte(body)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	var b bytes.Buffer
+	b.ReadFrom(resp.Body)
+	return resp.StatusCode, b.String()
+}
+
+func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) }

ai/internal/harness/client.go

@@ -57,6 +57,27 @@ type Proposal struct {
 	Approved  bool    `json:"approved"`
 }
 
+// ProposalState is GET /proposals/{token}: the decision as the harness knows it (approval may come through the API).
+type ProposalState struct {
+	Token      string  `json:"token"`
+	ActionID   string  `json:"action_id"`
+	Approved   bool    `json:"approved"`
+	ApprovedAt *string `json:"approved_at"`
+	ApprovedBy *string `json:"approved_by"`
+	Executed   bool    `json:"executed"`
+	ExecutedAt *string `json:"executed_at"`
+	ExpiresAt  string  `json:"expires_at"`
+}
+
+// GetProposal reads the harness-side state of a proposal.
+func (c *Client) GetProposal(ctx context.Context, token string) (*ProposalState, error) {
+	var out ProposalState
+	if _, err := c.call(ctx, "GET", "/proposals/"+token, nil, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
 // Propose registers a layer-2 proposal (POST /proposals); the harness records the proposal event and returns the token.
 func (c *Client) Propose(ctx context.Context, runID, scenarioID *string, actionID string, target, diagnosis map[string]any, ttl time.Duration) (*Proposal, error) {
 	body := map[string]any{"run_id": runID, "scenario_id": scenarioID, "action_id": actionID, "target": target, "diagnosis": diagnosis}

ai/internal/hitl/bot.go

@@ -230,12 +230,62 @@ func (b *Bot) edit(m *rag.HitlMessage, status string, color int, result map[stri
 	return b.dc.Edit(m.ChannelID, m.MessageID, em, b.buttons(m.Token, true))
 }
 
+// syncOne mirrors an approval recorded by the harness (API path, not the button) into the message. Returns true when
+// the message was decided here.
+func (b *Bot) syncOne(ctx context.Context, m *rag.HitlMessage) bool {
+	st, err := b.harness.GetProposal(ctx, m.Token)
+	if err != nil || !st.Approved {
+		return false
+	}
+	// Pending() carries no diagnosis; re-read the full row so the edited embed keeps label, evidence and rationale
+	// (campaign 5 C-G2, 2026-10-09: the mirrored message showed <nil> for every diagnosis field)
+	if full, err := b.store.GetHitl(ctx, m.Token); err == nil {
+		m = full
+	}
+	by := "unknown"
+	if st.ApprovedBy != nil && *st.ApprovedBy != "" {
+		by = *st.ApprovedBy
+	}
+	first, err := b.store.Decide(ctx, m.Token, "approved", by, map[string]any{"via": "harness-api", "executed": st.Executed})
+	if err != nil || !first {
+		return false
+	}
+	b.disarm(m.Token)
+	_ = b.edit(m, b.t.StatusApprovedAPI(by, st.Executed), colorApproved, nil)
+	log.Printf("proposal %s approved outside Discord by %s (executed=%v)", m.Token[len(m.Token)-6:], by, st.Executed)
+	return true
+}
+
+// SyncLoop polls the harness for pending proposals decided outside Discord (campaign 3/4 2026-10-08: approvals through
+// POST /proposals/{token}/approve left the message to expire by timer).
+func (b *Bot) SyncLoop(ctx context.Context, every time.Duration) {
+	t := time.NewTicker(every)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			list, err := b.store.Pending(ctx)
+			if err != nil {
+				continue
+			}
+			for i := range list {
+				b.syncOne(ctx, &list[i])
+			}
+		}
+	}
+}
+
 func (b *Bot) expire(token string) {
 	ctx := context.Background()
 	m, err := b.store.GetHitl(ctx, token)
 	if err != nil {
 		return
 	}
+	if b.syncOne(ctx, m) {
+		return
+	}
 	first, err := b.store.Decide(ctx, token, "expired", "timer", nil)
 	if err != nil || !first {
 		return

ai/internal/hitl/i18n.go

@@ -25,6 +25,7 @@ type Strings struct {
 	StatusExecFailed    func(by, err string) string
 	StatusExpiredBefore string
 	StatusNoDecision    string
+	StatusApprovedAPI   func(by string, executed bool) string
 	ExpiryMessage       func(actionID, target, runID, scenarioID string) string
 	EscalationTitle     func(label string) string
 	EscalationStatus    string
@@ -48,6 +49,12 @@ var langs = map[string]Strings{
 		StatusApproveFailed: func(e string) string { return "approval failed: " + e },
 		StatusExecFailed:    func(by, e string) string { return "approved by " + by + "; execution failed: " + e },
 		StatusExpiredBefore: "expired before approval",
+		StatusApprovedAPI: func(by string, executed bool) string {
+			if executed {
+				return "approved through the harness API by " + by + "; executed"
+			}
+			return "approved through the harness API by " + by
+		},
 		StatusNoDecision: "no decision within the window; moved to layer 3",
 		ExpiryMessage: func(a, t, r, s string) string {
 			return fmt.Sprintf("Proposal `%s` for `%s` (run %s, scenario %s) received no decision within the approval window. Layer 3: follow the runbook or call the harness `/actions/execute` with the operator token.", a, t, r, s)
@@ -72,6 +79,12 @@ var langs = map[string]Strings{
 		StatusApproveFailed: func(e string) string { return "承認の記録に失敗: " + e },
 		StatusExecFailed:    func(by, e string) string { return by + " が承認したが実行に失敗: " + e },
 		StatusExpiredBefore: "承認前に期限切れ",
+		StatusApprovedAPI: func(by string, executed bool) string {
+			if executed {
+				return by + " がハーネス API で承認、実行済み"
+			}
+			return by + " がハーネス API で承認"
+		},
 		StatusNoDecision: "期限内に判断がなかったため第3層へ引き継ぎ",
 		ExpiryMessage: func(a, t, r, s string) string {
 			return fmt.Sprintf("提案 `%s`（対象 `%s`、run %s、シナリオ %s）は承認期限内に判断されませんでした。第3層の対応です。ランブックに従って手動で対処するか、オペレーター用トークンでハーネスの `/actions/execute` を呼び出してください。", a, t, r, s)

ai/internal/tools/tools.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"net/url"
 	"os"
@@ -206,6 +207,24 @@ func (r *Runner) queryMetrics(ctx context.Context, promql string, minutes int64)
 	return map[string]any{"series_count": len(pr.Data.Result), "series": out, "range_minutes": minutes}, nil
 }
 
+// Instant runs an instant PromQL query and returns the first sample as a float (NaN when the series is absent).
+func (r *Runner) Instant(ctx context.Context, promql string) (float64, error) {
+	q := url.Values{"query": {promql}}
+	var pr promResp
+	if err := r.getJSON(ctx, strings.TrimRight(r.cfg.MimirURL, "/")+"/api/v1/query?"+q.Encode(), &pr); err != nil {
+		return math.NaN(), err
+	}
+	if len(pr.Data.Result) == 0 || len(pr.Data.Result[0].Value) != 2 {
+		return math.NaN(), nil
+	}
+	sv, _ := pr.Data.Result[0].Value[1].(string)
+	f, err := strconv.ParseFloat(sv, 64)
+	if err != nil {
+		return math.NaN(), err
+	}
+	return f, nil
+}
+
 func (r *Runner) probeStatus(ctx context.Context) (any, error) {
 	q := url.Values{"query": {"probe_success"}}
 	var pr promResp

harness/actions.go

@@ -127,6 +127,8 @@ func (a *App) authorize(ctx context.Context, req ActionRequest) (string, error)
 		return "", errors.New("invalid token")
 	}
 	switch {
+	case p.RejectedAt != nil:
+		return "", errors.New("proposal rejected")
 	case p.ApprovedAt == nil:
 		return "", errors.New("proposal not approved")
 	case p.ExecutedAt != nil:

harness/alerts.go

@@ -9,6 +9,8 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"sort"
+	"strings"
 	"sync"
 	"time"
 )
@@ -22,10 +24,41 @@ type AlertRouter struct {
 	app    *App
 	mu     sync.Mutex
 	timers map[string]*time.Timer // fingerprint -> escalation timer
+	firing map[string]string      // fingerprint -> alertname while Grafana reports it firing (judge: no recovery with an open alert)
+	// alertname+scenario -> when the last layer-2 diagnosis was dispatched. Grafana emits one alert per label set
+	// (ApiMemoryPressure: one per pod) and resends the group 30 s later; campaign 4 (2026-10-08) turned one fault into
+	// four diagnoses and four proposals. Inside cfg.AlertDedupeS the later alerts are recorded and skipped.
+	lastDiag map[string]time.Time
 }
 
 func NewAlertRouter(a *App) *AlertRouter {
-	return &AlertRouter{app: a, timers: map[string]*time.Timer{}}
+	return &AlertRouter{app: a, timers: map[string]*time.Timer{}, firing: map[string]string{}, lastDiag: map[string]time.Time{}}
+}
+
+// Firing lists the alert names currently firing (deduplicated, sorted).
+func (r *AlertRouter) Firing() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	seen := map[string]bool{}
+	var out []string
+	for _, n := range r.firing {
+		if !seen[n] {
+			seen[n] = true
+			out = append(out, n)
+		}
+	}
+	sort.Strings(out)
+	return out
+}
+
+func (r *AlertRouter) setFiring(fp, name string, on bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if on {
+		r.firing[fp] = name
+	} else {
+		delete(r.firing, fp)
+	}
 }
 
 // Grafana webhook payload (https://grafana.com/docs/grafana/latest/alerting/configure-notifications/manage-contact-points/integrations/webhook-notifier/)
@@ -80,8 +113,12 @@ func (r *AlertRouter) handleWebhook(w http.ResponseWriter, req *http.Request) {
 			"startsAt": al.StartsAt, "fingerprint": al.Fingerprint, "layer": al.Labels["layer"], "signature": sig}
 		r.app.store.Event(bg, runID, sid, "alert", "grafana", payload)
 		handled++
+		// layer-1 signature alerts are informational (the platform owns the recovery); they must not hold the judge
+		// (campaign 3, 2026-10-08: LegacyUnhealthy resolved 8.5 minutes after the container was back, V-K1 MTTR 631 s)
+		r.setFiring(al.Fingerprint, name, al.Status == "firing" && sig == "")
 		if al.Status != "firing" {
 			r.cancelTimer(al.Fingerprint)
+			r.clearDedupe(name, sid)
 			continue
 		}
 		if sig != "" {
@@ -98,11 +135,53 @@ func (r *AlertRouter) handleWebhook(w http.ResponseWriter, req *http.Request) {
 			r.startEscalation(al.Fingerprint, payload, runID, sid)
 			continue
 		}
+		if since, dup := r.dedupe(name, sid); dup {
+			r.app.store.Event(bg, runID, sid, "note", "runner", map[string]any{"phase": "dedupe", "alertname": name, "fingerprint": al.Fingerprint,
+				"since_last_diagnosis_s": since.Seconds(), "note": "same alert name already under diagnosis; no second /diagnose"})
+			continue
+		}
 		go r.diagnose(payload, runID, sid)
 	}
 	writeJSON(w, 200, map[string]any{"received": handled, "run_id": runID, "scenario_id": sid})
 }
 
+// dedupe reports whether a layer-2 diagnosis for this alert name (within the scenario) was dispatched less than
+// cfg.AlertDedupeS ago; when not, it records now as the dispatch time.
+func (r *AlertRouter) dedupe(name string, sid *string) (time.Duration, bool) {
+	if r.app.cfg.AlertDedupeS <= 0 {
+		return 0, false
+	}
+	key := name
+	if sid != nil {
+		key += "|" + *sid
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	now := time.Now()
+	if t, ok := r.lastDiag[key]; ok && now.Sub(t) < time.Duration(r.app.cfg.AlertDedupeS)*time.Second {
+		return now.Sub(t), true
+	}
+	r.lastDiag[key] = now
+	return 0, false
+}
+
+// clearDedupe forgets the dispatch time once no alert of that name is firing any more, so the next firing starts a
+// new diagnosis even inside the window.
+func (r *AlertRouter) clearDedupe(name string, sid *string) {
+	key := name
+	if sid != nil {
+		key += "|" + *sid
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, n := range r.firing {
+		if n == name {
+			return
+		}
+	}
+	delete(r.lastDiag, key)
+}
+
 func targetFromLabels(l map[string]string) map[string]any {
 	t := map[string]any{}
 	for _, k := range []string{"namespace", "pod", "deployment", "node", "instance"} {
@@ -148,11 +227,26 @@ func (r *AlertRouter) cancelTimer(fp string) {
 	}
 }
 
+// diagnoseTimeout bounds one /diagnose call: 3 minutes for API models (the AI VM gives up at 170 s); external-*
+// models (reference runs answered by a client outside the VM, apps 2026-10-16) get 26 minutes, one more than the AI
+// VM's EXTERNAL_DIAGNOSE_TIMEOUT_S default, so the VM's own error reaches the diagnosis event first.
+func diagnoseTimeout(model string) time.Duration {
+	if strings.HasPrefix(model, "external-") {
+		return 26 * time.Minute
+	}
+	return 3 * time.Minute
+}
+
 // diagnose asks the AI VM (design 6.3 strict schema). Until WP-12 the endpoint is absent; the failure is recorded as an event.
 func (r *AlertRouter) diagnose(alert map[string]any, runID, sid *string) {
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
-	defer cancel()
 	reqBody := map[string]any{"alert": alert, "run_id": runID, "scenario_id": sid}
+	model := ""
+	if cur := r.app.runs.current(); cur != nil && runID != nil && cur.RunID == *runID && cur.Model != "" {
+		model = cur.Model
+		reqBody["model"] = model // campaign-level model selection (the AI VM falls back to its default when absent)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), diagnoseTimeout(model))
+	defer cancel()
 	b, _ := json.Marshal(reqBody)
 	req, _ := http.NewRequestWithContext(ctx, "POST", r.app.cfg.AIDiagnoseURL, bytes.NewReader(b))
 	req.Header.Set("Content-Type", "application/json")

harness/alerts_test.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+// one fault, several alerts with the same name (per-pod label sets, group resend): one diagnosis inside the window
+func TestDedupeSameAlertName(t *testing.T) {
+	r := NewAlertRouter(&App{cfg: Config{AlertDedupeS: 300}})
+	sid := "C-G2"
+	if _, dup := r.dedupe("ApiMemoryPressure", &sid); dup {
+		t.Fatal("first alert must be diagnosed")
+	}
+	if _, dup := r.dedupe("ApiMemoryPressure", &sid); !dup {
+		t.Fatal("second alert inside the window must be skipped")
+	}
+	other := "V-G1"
+	if _, dup := r.dedupe("ApiMemoryPressure", &other); dup {
+		t.Fatal("another scenario is a new fault")
+	}
+	// all alerts of that name resolved -> the next firing starts a new diagnosis
+	r.setFiring("fp1", "ApiMemoryPressure", true)
+	r.clearDedupe("ApiMemoryPressure", &sid) // still firing: keep
+	if _, dup := r.dedupe("ApiMemoryPressure", &sid); !dup {
+		t.Fatal("still firing: keep the window")
+	}
+	r.setFiring("fp1", "ApiMemoryPressure", false)
+	r.clearDedupe("ApiMemoryPressure", &sid)
+	if _, dup := r.dedupe("ApiMemoryPressure", &sid); dup {
+		t.Fatal("after resolve the next firing is a new fault")
+	}
+	// window expiry
+	r.lastDiag["ApiMemoryPressure|"+sid] = time.Now().Add(-301 * time.Second)
+	if _, dup := r.dedupe("ApiMemoryPressure", &sid); dup {
+		t.Fatal("outside the window a new diagnosis is allowed")
+	}
+	// off
+	r2 := NewAlertRouter(&App{cfg: Config{AlertDedupeS: 0}})
+	if _, dup := r2.dedupe("X", nil); dup {
+		t.Fatal("dedupe off")
+	}
+	if _, dup := r2.dedupe("X", nil); dup {
+		t.Fatal("dedupe off")
+	}
+}
+
+// external-* models (reference runs served by a client outside the AI VM) get the long diagnose timeout; the model id
+// itself passes the run request bound.
+func TestDiagnoseTimeoutByModel(t *testing.T) {
+	if d := diagnoseTimeout("claude-sonnet-5-5"); d != 3*time.Minute {
+		t.Fatalf("api model timeout %v", d)
+	}
+	if d := diagnoseTimeout(""); d != 3*time.Minute {
+		t.Fatalf("default model timeout %v", d)
+	}
+	if d := diagnoseTimeout("external-claude-mythos-5-1"); d != 26*time.Minute {
+		t.Fatalf("external model timeout %v", d)
+	}
+	if !modelRe.MatchString("external-claude-mythos-5-1") {
+		t.Fatal("external model id must pass modelRe")
+	}
+}

harness/autoapprove.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"time"
+)
+
+// autoApprove is the measurement-mode stand-in for the Discord approver (run option auto_approve_after_s, 2026-10-15):
+// after the configured delay the harness approves the layer-2 proposal (actor "auto") and executes it exactly as the
+// HITL bot does after a button press (actor "ai-approved", MarkProposalExecuted). MTTA therefore equals the delay;
+// the verdict logic is unchanged (layer 2 from the ai-approved action). A proposal already decided, expired or
+// recorded-only is handled the same way as the API path. The claim (approve + executed_at) is a single conditional
+// UPDATE, so a human approval/execution or a rejection that arrives first takes precedence (no double execution).
+func (a *App) autoApprove(token string, req ProposalRequest, after time.Duration) {
+	time.Sleep(after)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+	defer cancel()
+	p, err := a.store.GetProposal(ctx, token)
+	if err != nil {
+		return
+	}
+	// one conditional update claims approval + execution: a human/bot decision (approve, reject, execute) that lands
+	// first wins and this goroutine does nothing
+	claimed, err := a.store.ClaimProposal(ctx, token, "auto")
+	if err != nil || !claimed {
+		return
+	}
+	suffix := token[len(token)-6:]
+	a.store.Event(ctx, p.RunID, p.ScenarioID, "approval", "auto", map[string]any{"action_id": p.ActionID, "token_suffix": suffix, "auto_approve_after_s": after.Seconds()})
+	ce := catalogEntry(p.ActionID)
+	if ce == nil {
+		return
+	}
+	note := "auto-approved by the harness (measurement mode)"
+	if !ce.Exec {
+		a.store.Event(ctx, p.RunID, p.ScenarioID, "action", "ai-approved", map[string]any{"action_id": p.ActionID, "target": req.Target, "recorded_only": true, "note": note, "layer": ce.Layer})
+		mActs.WithLabelValues(p.ActionID, "ai-approved", "recorded").Inc()
+		return
+	}
+	res, err := a.executeAction(ctx, p.ActionID, req.Target, "ai-approved", p.RunID, p.ScenarioID)
+	result := "ok"
+	if err != nil {
+		result = "error"
+	}
+	mActs.WithLabelValues(p.ActionID, "ai-approved", result).Inc()
+	a.store.Event(ctx, p.RunID, p.ScenarioID, "action", "ai-approved", map[string]any{"action_id": p.ActionID, "target": req.Target, "result": res, "error": errStr(err), "note": note})
+}

harness/gcp.go

@@ -60,6 +60,9 @@ func (a *App) gcpListInstances(ctx context.Context, prefix string) ([]string, er
 
 // resolveInstance turns "legacy:N" (N-th MIG instance by name) or a pet VM short name (db-primary -> selfheal-db-primary) into an instance.
 func (a *App) resolveInstance(ctx context.Context, ref string) (gceInstance, error) {
+	if ref == "legacy:busy" {
+		return a.resolveBusyLegacy(ctx)
+	}
 	if strings.HasPrefix(ref, "legacy:") {
 		n, _ := strconv.Atoi(strings.TrimPrefix(ref, "legacy:"))
 		l, err := a.gcpInstances(ctx, "legacy-")
@@ -170,3 +173,36 @@ func (a *App) gcpOp(ctx context.Context, op, instance, mig string, size int64) (
 
 func (a *App) region(zone string) string { return zone[:strings.LastIndex(zone, "-")] }
 func (a *App) regionOfMIG() string       { return env("GCP_REGION", "asia-northeast1") }
+
+// resolveBusyLegacy picks the MIG instance that served the most legacy-api requests over the last 2 minutes. The ILB
+// hashes the api pods' few keep-alive connections onto one instance (campaign 2, 2026-10-08: legacy:0 served 0 rps while
+// the delay was on it), so a fault must go where the traffic is. The choice is cached per run+scenario so the revert
+// hits the same instance after the traffic has moved away from it.
+func (a *App) resolveBusyLegacy(ctx context.Context) (gceInstance, error) {
+	key := ""
+	if cur := a.runs.current(); cur != nil {
+		key = cur.RunID + "/" + cur.ScenarioID
+	}
+	if key != "" {
+		if v, ok := a.busyLegacy.Load(key); ok {
+			return v.(gceInstance), nil
+		}
+	}
+	name, err := a.mimirTopInstance(ctx, `topk(1, sum by (instance) (rate(legacy_http_requests_total{route!="healthz"}[2m])))`)
+	if err != nil {
+		return gceInstance{}, fmt.Errorf("legacy:busy: %w", err)
+	}
+	l, err := a.gcpInstances(ctx, "legacy-")
+	if err != nil {
+		return gceInstance{}, err
+	}
+	for _, i := range l {
+		if i.Name == name {
+			if key != "" {
+				a.busyLegacy.Store(key, i)
+			}
+			return i, nil
+		}
+	}
+	return gceInstance{}, fmt.Errorf("legacy:busy: instance %q from Mimir not in the MIG (%d instances)", name, len(l))
+}

harness/judge.go

@@ -45,16 +45,51 @@ func (a *App) mimirScalar(ctx context.Context, expr string) (float64, error) {
 	return strconv.ParseFloat(s, 64)
 }
 
+// mimirTopInstance runs an instant query and returns the "instance" label of the first sample.
+func (a *App) mimirTopInstance(ctx context.Context, expr string) (string, error) {
+	u := a.cfg.MimirURL + "/api/v1/query?query=" + url.QueryEscape(expr)
+	req, _ := http.NewRequestWithContext(ctx, "GET", u, nil)
+	resp, err := (&http.Client{Timeout: 15 * time.Second}).Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	b, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != 200 {
+		return "", fmt.Errorf("mimir %s: %s", resp.Status, truncate(string(b), 300))
+	}
+	var r struct {
+		Data struct {
+			Result []struct {
+				Metric map[string]string `json:"metric"`
+			} `json:"result"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(b, &r); err != nil {
+		return "", err
+	}
+	if len(r.Data.Result) == 0 || r.Data.Result[0].Metric["instance"] == "" {
+		return "", fmt.Errorf("no instance sample for %s", expr)
+	}
+	return r.Data.Result[0].Metric["instance"], nil
+}
+
 // SLOState is one evaluation of experiment design 4.2 (probe 12 consecutive successes = min over 3m; p95 and 5xx ratio over 5m).
 // Values are nil when the series is absent (JSON has no NaN).
+// 2026-10-08 (campaign 2, C-G1): a 300 ms delay that lasted 3 minutes was declared recovered while still active because the
+// 5-minute p95 average diluted it to 0.375 s. The 1-minute p95 (p95_1m_s) must also be inside the SLO.
 type SLOState struct {
 	At           time.Time `json:"at"`
 	ProbeMin3m   *float64  `json:"probe_min_3m"`
 	P95          *float64  `json:"p95_s"`
+	P95Now       *float64  `json:"p95_1m_s"`
 	ErrorRatio   *float64  `json:"error_ratio"`
 	ProbeOK      bool      `json:"probe_ok"`
 	LatencyOK    bool      `json:"latency_ok"`
+	LatencyNowOK bool      `json:"latency_1m_ok"`
 	ErrorsOK     bool      `json:"errors_ok"`
+	AlertsFiring []string  `json:"alerts_firing,omitempty"` // Grafana alerts received firing and not yet resolved (2026-10-08: grey-zone symptoms live outside the three SLO series)
+	AlertsOK     bool      `json:"alerts_ok"`
 	Recovered    bool      `json:"recovered"`
 	MissingData  []string  `json:"missing_data,omitempty"`
 	QueryErrors  []string  `json:"query_errors,omitempty"`
@@ -63,6 +98,7 @@ type SLOState struct {
 const (
 	qProbeMin3m = `min_over_time(probe_success{job="blackbox",target="alb"}[3m])`
 	qP95        = `histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket{route=~"orders|price"}[5m])))`
+	qP95Now     = `histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket{route=~"orders|price"}[1m])))`
 	qErrRatio   = `sum(rate(demo_http_requests_total{route=~"orders|price",status=~"5.."}[5m])) / sum(rate(demo_http_requests_total{route=~"orders|price"}[5m]))`
 )
 
@@ -84,8 +120,14 @@ func (a *App) evaluateSLO(ctx context.Context) SLOState {
 	st.ProbeOK = st.ProbeMin3m != nil && *st.ProbeMin3m >= 1 // the probe series must exist: no data is not recovery
 	st.P95 = get("p95", qP95)
 	st.LatencyOK = st.P95 == nil || *st.P95 < 0.5 // no traffic in the window (no k6) counts as within SLO
+	st.P95Now = get("p95_1m", qP95Now)
+	st.LatencyNowOK = st.P95Now == nil || *st.P95Now < 0.5
 	st.ErrorRatio = get("errors", qErrRatio)
 	st.ErrorsOK = st.ErrorRatio == nil || *st.ErrorRatio < 0.01
-	st.Recovered = st.ProbeOK && st.LatencyOK && st.ErrorsOK && len(st.QueryErrors) == 0
+	if a.hooks != nil {
+		st.AlertsFiring = a.hooks.Firing()
+	}
+	st.AlertsOK = len(st.AlertsFiring) == 0
+	st.Recovered = st.ProbeOK && st.LatencyOK && st.LatencyNowOK && st.ErrorsOK && st.AlertsOK && len(st.QueryErrors) == 0
 	return st
 }

harness/main.go

@@ -9,7 +9,9 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -32,6 +34,7 @@ type Config struct {
 	DBPassFile    string // harness role password on db-primary / db-replica
 	HumanTokFile  string
 	AIDiagnoseURL string
+	AlertDedupeS  int // layer-2 alerts with the same name inside this window share one diagnosis (0 = off)
 	ScenarioDir   string
 	JudgeEvery    time.Duration
 	ObserveMax    time.Duration
@@ -61,6 +64,7 @@ func loadConfig() Config {
 		DBPassFile:    env("DB_PASSWORD_FILE", "/run/secrets/db-password"),
 		HumanTokFile:  env("HUMAN_TOKEN_FILE", "/run/secrets/harness-token"),
 		AIDiagnoseURL: env("AI_DIAGNOSE_URL", "http://10.20.0.12:8080/diagnose"),
+		AlertDedupeS:  envInt("ALERT_DEDUPE_S", 300),
 		ScenarioDir:   env("SCENARIO_DIR", "/etc/harness/scenarios"),
 		JudgeEvery:    30 * time.Second,
 		ObserveMax:    30 * time.Minute,
@@ -79,6 +83,7 @@ type App struct {
 	scen       map[string]*Scenario
 	runs       *RunManager
 	hooks      *AlertRouter
+	busyLegacy sync.Map // run/scenario -> gceInstance chosen for legacy:busy (gcp.go)
 }
 
 func main() {
@@ -113,6 +118,7 @@ func main() {
 	mux.HandleFunc("POST /runs/{id}/abort", app.handleRunAbort)
 	mux.HandleFunc("POST /hooks/alert", app.hooks.handleWebhook)
 	mux.HandleFunc("POST /proposals", app.handleProposal)
+	mux.HandleFunc("GET /proposals/{token}", app.handleProposalGet)
 	mux.HandleFunc("POST /proposals/{token}/approve", app.handleProposalApprove)
 	mux.HandleFunc("POST /proposals/{token}/reject", app.handleProposalReject)
 	mux.HandleFunc("POST /actions/execute", app.handleActionExecute)
@@ -206,3 +212,12 @@ func (a *App) handleScenarios(w http.ResponseWriter, r *http.Request) {
 	sortScenarios(list)
 	writeJSON(w, 200, list)
 }
+
+func envInt(k string, d int) int {
+	if v := os.Getenv(k); v != "" {
+		if n, err := strconv.Atoi(v); err == nil {
+			return n
+		}
+	}
+	return d
+}

harness/proposals.go

@@ -59,9 +59,24 @@ func (a *App) handleProposal(w http.ResponseWriter, r *http.Request) {
 		_ = a.store.ApproveProposal(r.Context(), token, by)
 		a.store.Event(r.Context(), req.RunID, req.ScenarioID, "approval", by, map[string]any{"action_id": req.ActionID, "token_suffix": token[len(token)-6:]})
 	}
+	if cur := a.runs.current(); !req.Approved && cur != nil && cur.AutoApproveAfter > 0 && req.RunID != nil && *req.RunID == cur.RunID {
+		go a.autoApprove(token, req, cur.AutoApproveAfter) // measurement mode (run option auto_approve_after_s)
+	}
 	writeJSON(w, 201, map[string]any{"token": token, "expires_in_s": ttl.Seconds(), "approved": req.Approved})
 }
 
+// GET /proposals/{token}: the HITL bot reads the decision state (approval may arrive through this API instead of the
+// Discord button; the bot mirrors it into its message instead of letting the message expire, campaign 3/4 2026-10-08).
+func (a *App) handleProposalGet(w http.ResponseWriter, r *http.Request) {
+	p, err := a.store.GetProposal(r.Context(), r.PathValue("token"))
+	if err != nil {
+		writeErr(w, 404, errors.New("unknown proposal"))
+		return
+	}
+	writeJSON(w, 200, map[string]any{"token": p.Token, "run_id": p.RunID, "scenario_id": p.ScenarioID, "action_id": p.ActionID,
+		"approved": p.ApprovedAt != nil, "approved_at": p.ApprovedAt, "approved_by": p.ApprovedBy, "executed": p.ExecutedAt != nil, "executed_at": p.ExecutedAt, "expires_at": p.ExpiresAt})
+}
+
 // POST /proposals/{token}/approve: the HITL bot records the human decision (design 6.5 step 4). The proposal was
 // registered at diagnosis time, so MTTA = approval - proposal (design 5.4) measures the real waiting time.
 func (a *App) handleProposalApprove(w http.ResponseWriter, r *http.Request) {
@@ -79,6 +94,9 @@ func (a *App) handleProposalApprove(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	switch {
+	case p.RejectedAt != nil:
+		writeErr(w, 409, errors.New("proposal rejected"))
+		return
 	case p.ApprovedAt != nil:
 		writeErr(w, 409, errors.New("proposal already approved"))
 		return
@@ -107,6 +125,7 @@ func (a *App) handleProposalReject(w http.ResponseWriter, r *http.Request) {
 		writeErr(w, 404, errors.New("unknown proposal"))
 		return
 	}
+	_ = a.store.RejectProposal(r.Context(), token) // blocks a later approval (button, API or auto-approval)
 	a.store.Event(r.Context(), p.RunID, p.ScenarioID, "note", req.RejectedBy, map[string]any{"rejected": true, "action_id": p.ActionID, "reason": req.Reason, "token_suffix": token[len(token)-6:]})
 	writeJSON(w, 200, map[string]any{"token": token, "rejected": true})
 }

harness/run.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
@@ -25,6 +26,8 @@ type RunManager struct {
 type activeRun struct {
 	RunID            string
 	Campaign         string
+	Model            string        // LLM for this run's diagnoses ("" = AI VM default); passed to the AI VM on every /diagnose
+	AutoApproveAfter time.Duration // >0: measurement mode, the harness approves and executes layer-2 proposals after this delay
 	ScenarioID       string        // "" between scenarios
 	Scenario         *Scenario
 	TInject          time.Time
@@ -33,6 +36,7 @@ type activeRun struct {
 
 type RunRequest struct {
 	Campaign          string   `json:"campaign"`
+	Model             string   `json:"model"` // optional: LLM id for the diagnoses (e.g. gpt-6-luna, claude-sonnet-5-5); "" = AI VM default
 	ScenarioIDs       []string `json:"scenario_ids"`
 	BaselineS         *int     `json:"baseline_s"`        // default 300 (design 8)
 	CooldownS         *int     `json:"cooldown_s"`        // default 300
@@ -42,8 +46,13 @@ type RunRequest struct {
 	GitSHA            string   `json:"git_sha"`
 	Notes             string   `json:"notes"`
 	SkipRevert        bool     `json:"skip_revert"`
+	AutoApproveAfterS *int     `json:"auto_approve_after_s"` // measurement mode (model comparison without a human in the loop): nil/0 = off
 }
 
+// modelRe bounds the model id (empty = AI VM default). The AI VM routes by prefix: claude-* -> Anthropic,
+// external-* -> external responder (reference runs, longer diagnose timeout in alerts.go), else OpenAI.
+var modelRe = regexp.MustCompile(`^[A-Za-z0-9._:-]{0,64}$`)
+
 func NewRunManager(a *App) *RunManager { return &RunManager{app: a} }
 
 func (m *RunManager) current() *activeRun {
@@ -78,6 +87,10 @@ func (a *App) handleRunCreate(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	}
+	if !modelRe.MatchString(req.Model) {
+		writeErr(w, 400, errors.New("model must match [A-Za-z0-9._:-]{1,64}"))
+		return
+	}
 	m := a.runs
 	m.mu.Lock()
 	if m.active != nil {
@@ -88,9 +101,13 @@ func (a *App) handleRunCreate(w http.ResponseWriter, r *http.Request) {
 	}
 	runID := newID("run-")
 	ctx, cancel := context.WithCancel(context.Background())
-	m.active = &activeRun{RunID: runID, Campaign: req.Campaign, cancel: cancel}
+	var autoAfter time.Duration
+	if req.AutoApproveAfterS != nil && *req.AutoApproveAfterS > 0 {
+		autoAfter = time.Duration(*req.AutoApproveAfterS) * time.Second
+	}
+	m.active = &activeRun{RunID: runID, Campaign: req.Campaign, Model: req.Model, AutoApproveAfter: autoAfter, cancel: cancel}
 	m.mu.Unlock()
-	if err := a.store.CreateRun(r.Context(), runID, req.Campaign, req.DriftBefore, req.GitSHA, req.Notes); err != nil {
+	if err := a.store.CreateRun(r.Context(), runID, req.Campaign, req.DriftBefore, req.GitSHA, req.Notes, req.Model, req.AutoApproveAfterS); err != nil {
 		m.mu.Lock()
 		m.active = nil
 		m.mu.Unlock()
@@ -100,7 +117,7 @@ func (a *App) handleRunCreate(w http.ResponseWriter, r *http.Request) {
 	}
 	mRuns.Set(1)
 	go m.execute(ctx, runID, req)
-	writeJSON(w, 202, map[string]any{"run_id": runID, "scenario_ids": req.ScenarioIDs})
+	writeJSON(w, 202, map[string]any{"run_id": runID, "scenario_ids": req.ScenarioIDs, "model": req.Model, "auto_approve_after_s": req.AutoApproveAfterS})
 }
 
 func (a *App) handleRunAbort(w http.ResponseWriter, r *http.Request) {
@@ -262,11 +279,22 @@ func (m *RunManager) runScenario(ctx context.Context, runID string, sc *Scenario
 	aborted := ctx.Err() != nil
 	// revert (always, even on abort, unless the scenario is irreversible by design or skip_revert)
 	if !skipRevert && sc.Revert.Method != "none" {
+		// a failed revert leaves the fault in place (2026-10-18: one SSH connection reset at the X-M2 revert left a
+		// netem qdisc on a legacy instance and broke the next 14 runs of two scenarios), so retry a few times
+		var rres map[string]any
+		var rerr error
+		for attempt := 1; attempt <= 3; attempt++ {
 			rvCtx, cancelRv := context.WithTimeout(bg, 10*time.Minute)
-		rres, rerr := a.runStep(rvCtx, sc, sc.Revert, runID, params)
+			rres, rerr = a.runStep(rvCtx, sc, sc.Revert, runID, params)
 			cancelRv()
+			if rerr == nil {
+				break
+			}
+			a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "revert", "attempt": attempt, "error": rerr.Error(), "result": rres})
+			time.Sleep(15 * time.Second)
+		}
 		if rerr != nil {
-			a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "revert", "error": rerr.Error(), "result": rres})
+			a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "revert", "error": rerr.Error(), "result": rres, "note": "revert failed after 3 attempts; the fault may still be in place"})
 		} else {
 			a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "revert", "result": rres})
 		}

harness/store.go

@@ -24,6 +24,9 @@ CREATE TABLE IF NOT EXISTS runs (
   status       text NOT NULL DEFAULT 'running',
   notes        text
 );
+ALTER TABLE runs ADD COLUMN IF NOT EXISTS model text;  -- LLM requested for the run's diagnoses (campaign-level model selection, 2026-10-14)
+ALTER TABLE runs ADD COLUMN IF NOT EXISTS auto_approve_after_s int;  -- measurement mode: the harness approves + executes layer-2 proposals after this delay (2026-10-15)
+ALTER TABLE proposals ADD COLUMN IF NOT EXISTS rejected_at timestamptz;  -- POST /proposals/{token}/reject (a rejected proposal can no longer be approved or executed)
 CREATE TABLE IF NOT EXISTS scenarios (
   scenario_id         text PRIMARY KEY,
   path                text NOT NULL,
@@ -119,8 +122,8 @@ func (s *Store) UpsertScenarios(ctx context.Context, m map[string]*Scenario) err
 	return nil
 }
 
-func (s *Store) CreateRun(ctx context.Context, id, campaign string, driftBefore *int, gitSHA, notes string) error {
+func (s *Store) CreateRun(ctx context.Context, id, campaign string, driftBefore *int, gitSHA, notes, model string, autoApproveAfterS *int) error {
-	_, err := s.pool.Exec(ctx, `INSERT INTO runs (run_id,campaign,drift_before,git_sha,notes) VALUES ($1,$2,$3,$4,$5)`, id, campaign, driftBefore, nz(gitSHA), nz(notes))
+	_, err := s.pool.Exec(ctx, `INSERT INTO runs (run_id,campaign,drift_before,git_sha,notes,model,auto_approve_after_s) VALUES ($1,$2,$3,$4,$5,$6,$7)`, id, campaign, driftBefore, nz(gitSHA), nz(notes), nz(model), autoApproveAfterS)
 	return err
 }
 
@@ -222,10 +225,12 @@ type RunRow struct {
 	GitSHA      *string    `json:"git_sha"`
 	Status      string     `json:"status"`
 	Notes       *string    `json:"notes"`
+	Model       *string    `json:"model"`                // nil for runs before campaign-level model selection
+	AutoApprove *int       `json:"auto_approve_after_s"` // nil/0 = human approval (Discord); >0 = harness auto-approved layer-2 proposals after this delay
 }
 
 func (s *Store) Runs(ctx context.Context, runID string) ([]RunRow, error) {
-	rows, err := s.pool.Query(ctx, `SELECT run_id,campaign,started_at,finished_at,drift_before,drift_after,git_sha,status,notes FROM runs WHERE ($1='' OR run_id=$1) ORDER BY started_at DESC LIMIT 100`, runID)
+	rows, err := s.pool.Query(ctx, `SELECT run_id,campaign,started_at,finished_at,drift_before,drift_after,git_sha,status,notes,model,auto_approve_after_s FROM runs WHERE ($1='' OR run_id=$1) ORDER BY started_at DESC LIMIT 100`, runID)
 	if err != nil {
 		return nil, err
 	}
@@ -233,7 +238,7 @@ func (s *Store) Runs(ctx context.Context, runID string) ([]RunRow, error) {
 	var out []RunRow
 	for rows.Next() {
 		var r RunRow
-		if err := rows.Scan(&r.RunID, &r.Campaign, &r.StartedAt, &r.FinishedAt, &r.DriftBefore, &r.DriftAfter, &r.GitSHA, &r.Status, &r.Notes); err != nil {
+		if err := rows.Scan(&r.RunID, &r.Campaign, &r.StartedAt, &r.FinishedAt, &r.DriftBefore, &r.DriftAfter, &r.GitSHA, &r.Status, &r.Notes, &r.Model, &r.AutoApprove); err != nil {
 			return nil, err
 		}
 		out = append(out, r)
@@ -283,13 +288,14 @@ type ProposalRow struct {
 	ApprovedAt *time.Time
 	ApprovedBy *string
 	ExecutedAt *time.Time
+	RejectedAt *time.Time
 	ExpiresAt  time.Time
 }
 
 func (s *Store) GetProposal(ctx context.Context, token string) (*ProposalRow, error) {
 	var p ProposalRow
-	err := s.pool.QueryRow(ctx, `SELECT token,run_id,scenario_id,action_id,approved_at,approved_by,executed_at,expires_at FROM proposals WHERE token=$1`, token).
+	err := s.pool.QueryRow(ctx, `SELECT token,run_id,scenario_id,action_id,approved_at,approved_by,executed_at,rejected_at,expires_at FROM proposals WHERE token=$1`, token).
-		Scan(&p.Token, &p.RunID, &p.ScenarioID, &p.ActionID, &p.ApprovedAt, &p.ApprovedBy, &p.ExecutedAt, &p.ExpiresAt)
+		Scan(&p.Token, &p.RunID, &p.ScenarioID, &p.ActionID, &p.ApprovedAt, &p.ApprovedBy, &p.ExecutedAt, &p.RejectedAt, &p.ExpiresAt)
 	if err != nil {
 		return nil, err
 	}
@@ -297,7 +303,23 @@ func (s *Store) GetProposal(ctx context.Context, token string) (*ProposalRow, er
 }
 
 func (s *Store) ApproveProposal(ctx context.Context, token, by string) error {
-	_, err := s.pool.Exec(ctx, `UPDATE proposals SET approved_at=now(), approved_by=$2 WHERE token=$1 AND approved_at IS NULL`, token, by)
+	_, err := s.pool.Exec(ctx, `UPDATE proposals SET approved_at=now(), approved_by=$2 WHERE token=$1 AND approved_at IS NULL AND rejected_at IS NULL`, token, by)
+	return err
+}
+
+// ClaimProposal approves and marks executed in one conditional update, so exactly one caller (auto-approval or a
+// human/bot execute racing with it) wins: it reports whether this call claimed the proposal.
+func (s *Store) ClaimProposal(ctx context.Context, token, by string) (bool, error) {
+	tag, err := s.pool.Exec(ctx, `UPDATE proposals SET approved_at=now(), approved_by=$2, executed_at=now()
+		WHERE token=$1 AND approved_at IS NULL AND executed_at IS NULL AND rejected_at IS NULL AND expires_at > now()`, token, by)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+func (s *Store) RejectProposal(ctx context.Context, token string) error {
+	_, err := s.pool.Exec(ctx, `UPDATE proposals SET rejected_at=now() WHERE token=$1 AND approved_at IS NULL AND rejected_at IS NULL`, token)
 	return err
 }