[v2] 模型请求被限值切断后，Agent 不再反复重发，这一轮显示切断原因

frontend-v2/src/components/agent-thread/model/history.test.ts

@@ -200,6 +200,15 @@ describe('conversationOf', () => {
     expect(turn.error).toContain('history_limit')
   })
 
+  it('says which model limit stopped a turn that Node only recorded as cancelled', () => {
+    for (const [code, text] of [['model_max_duration', '超过时长上限'], ['model_idle_timeout', '长时间没有输出']]) {
+      const failed: ChatResult = { status: 'failed', answer: '', answerComplete: false, stopReason: 'cancelled', failureCode: code }
+      const [turn] = conversationOf([view({ records: [user(1, 'a'), say(2, 'Start. '), ended(3, 'cancelled')], results: [failed] })]).turns
+      expect(turn.status).toBe('failed')
+      expect(turn.error).toContain(text)
+    }
+  })
+
   it('marks where an ended session stops the conversation, as interrupted when chat did not end it', () => {
     const stopped = conversationOf([view({ id: 's1', records: [user(1, 'a'), ended(2)], results: [done], ended: true, endReason: 'stopped' })])
     expect(stopped.markers).toEqual([{ kind: 'session', id: 'session-s1', afterTurn: 1, reason: 'stopped' }])

frontend-v2/src/components/agent-thread/model/history.ts

@@ -214,8 +214,15 @@ function markFailures(conversation: Conversation, from: number, results: ChatRes
   return changed ? { ...conversation, turns } : conversation
 }
 
+/** Model limits that v2-server enforces on one call; the Agent itself only reports a cancel. */
+const limitTexts: Record<string, string> = {
+  model_max_duration: '模型请求超过时长上限被切断，这一轮已停止。',
+  model_idle_timeout: '模型长时间没有输出被切断，这一轮已停止。',
+}
+
 function failureText(result: ChatResult): string {
   if (result.status === 'interrupted') return '服务重启，这一轮被中断。'
+  if (limitTexts[result.failureCode]) return limitTexts[result.failureCode]
   if (!result.answerComplete) return `回答没能完整读取${result.failureCode ? `（${result.failureCode}）` : ''}，这一轮按失败处理。`
   return `这一轮失败了${result.failureCode ? `（${result.failureCode}）` : ''}。`
 }

v2-server/README.md

@@ -35,7 +35,7 @@ go run .                               # 监听 127.0.0.1:8090
 | `V2_NODE_ASSETS_DIR` | 未设置 | `task v2:node:prepare` 生成的固定 Node、host、guardian、Deno、OpenCode 和 Claude Code 资源；未设置时只启用纯模型路径 |
 | `V2_MODEL_BASE_URL` / `V2_MODEL_API_KEY` / `V2_MODEL_NAME` | 未设置 | 可选的一组 HTTPS OpenAI 兼容 provider；必须同时设置，OpenCode 与纯模型共用，清单只公开模型名称与版本 |
 | `V2_ANTHROPIC_MODEL_BASE_URL` / `V2_ANTHROPIC_MODEL_API_KEY` / `V2_ANTHROPIC_MODEL_NAME` | 未设置 | 可选的一组 HTTPS Anthropic 兼容 provider（Messages API，base URL 写到 `/v1`，如 `https://api.anthropic.com/v1`）；必须同时提供地址、凭据和模型名，只供 Claude Code 使用。使用 Bearer 网关时以 `V2_ANTHROPIC_MODEL_AUTH_TOKEN` 代替 `V2_ANTHROPIC_MODEL_API_KEY`，二者不能同时配置；Claude Code 的 base URL 若以 `/compatible` 结尾，这里填 `/compatible/v1` |
-| `V2_MODEL_IDLE_TIMEOUT` / `V2_MODEL_MAX_DURATION` | `2m` / `30m` | 单次上游模型请求的空闲超时和总上限，两种协议共用；必须为正 duration，空闲超时不能大于总上限。连续这么久收不到上游数据、或总时长到上限时中止响应，Agent 收到的是失败而不是正常结束的流 |
+| `V2_MODEL_IDLE_TIMEOUT` / `V2_MODEL_MAX_DURATION` | `2m` / `30m` | 单次上游模型请求的空闲超时和总上限，两种协议共用；必须为正 duration，空闲超时不能大于总上限。连续这么久收不到上游数据、或总时长到上限时中止响应，Agent 收到的是失败而不是正常结束的流；总上限切断后这一轮不再放行重发，空闲超时切断后只放行 1 次，这一轮按失败结束并显示原因 |
 
 测试需要 `TEST_DATABASE_URL`（任意可建 schema 的库），每个测试用独立的 schema：
 

v2-server/internal/agentruntime/README.md

@@ -38,6 +38,10 @@ err = driver.Close(ctx)
 
 End 等待当前 Node incarnation 的有效 `agent_session_ended`；缺字段、未知原因、错 Node / incarnation 或传输断开不能当成停止确认。项目已 dead/fatal 时，End 使用调用方 context 走 supervisor 清理；确认旧进程树停止并完成关闭后移除原项目条目，不需要另一个 Run 触发回收。停止失败或超时返回 `ErrStopUnconfirmed`，保留原项目，供后续 End 重试；上层不能释放目录或派发后继。Close 最后通过已有 Node supervisor 确认所有自有进程退出；失败可重试。目录和工具写入在正常 End/Close 后保留。
 
+## 被切断后的重发
+
+Agent 把被切断的模型请求当成网络失败原样重发（OpenCode 0.6.11 最多 5 次，Claude Code 0.4.5 十次以上），每次都把半截回答再流一遍。每个模型请求都套上所在轮的守卫：总上限切断后，这一轮之后的请求都在响应头前用 `RefuseCall` 拒绝；空闲超时切断后放行 1 次重发，再被空闲超时切断就同样拒绝。响应头之前和之后到时的限值一样计。同时在跑、一起被切断的请求（例如标题请求）只算一次；切断之后开始、上游 2xx 且没被切断的请求清零，其它失败（上游拒绝、连不上）不改变计数，Agent 对它们的重发归它自己的策略。Node 把失败的轮次记成 `cancelled`、不带错误文字，所以这一轮最后的请求是被限值切断的时候，结果改为失败，`FailureCode` 就是切断原因，供页面显示。
+
 ## 首次输出前的等待
 
 会话第一轮到首个可见输出之前的等待按五段计时：`runtime`（启动项目 Node）、`workspace`（准备工作目录和技能，用上预热副本时标 `reused`）、`plugin`（等插件安装剩余部分，安装与 clone 并行，这里只算 clone 之后还要等的时间）、`agent`（从发出 `start_agent_session` 到 Node 记录会话）、`response`（从会话记录到首个思考、回答或工具调用）。每段接着上一段开始，加起来就是这段等待。

v2-server/internal/agentruntime/project.go

@@ -235,26 +235,31 @@ func (p *projectRuntime) start() error {
 
 // providerFor binds each observed HTTP request to the current admitted turn, not a browser ID.
 // Returning nil denies a request when its execution or active turn is no longer present.
+// Every call of an active turn also passes the turn's retry guard.
 func (p *projectRuntime) providerFor(id string, binding nodeservices.Binding) nodeservices.Provider {
 	observe := p.driver.opts.ObserveProvider
-	if observe == nil {
-		return binding.Provider
-	}
 	p.mu.Lock()
 	a := p.agents[id]
 	p.mu.Unlock()
 	if a == nil {
+		if observe == nil {
+			return binding.Provider
+		}
 		return nil
 	}
-	if !a.observeRequests {
-		return binding.Provider
-	}
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	if a.active == nil {
+		if observe == nil || !a.observeRequests {
+			return binding.Provider
+		}
 		return nil
 	}
-	return observe(execution.Ref{SessionID: a.frozen.ID, TurnID: a.active.id}, binding)
+	provider := binding.Provider
+	if observe != nil && a.observeRequests {
+		provider = observe(execution.Ref{SessionID: a.frozen.ID, TurnID: a.active.id}, binding)
+	}
+	return a.active.guard.provider(provider, binding.Protocol)
 }
 
 func (p *projectRuntime) ipc() nodeClient {

v2-server/internal/agentruntime/provider_test.go

@@ -31,8 +31,9 @@ func TestProviderForCurrentObservedTurnAndPassthrough(t *testing.T) {
 	}
 	for _, turn := range []string{"one", "two"} {
 		a.active = &activeTurn{id: turn}
-		if got := p.providerFor("execution", binding); !reflect.DeepEqual(got, observed) {
-			t.Fatal("observer not wired")
+		got, ok := p.providerFor("execution", binding).(guardedProvider)
+		if !ok || !reflect.DeepEqual(got.next, observed) || got.guard != &a.active.guard {
+			t.Fatal("observer not wired behind the turn's retry guard")
 		}
 	}
 	if !reflect.DeepEqual(refs, []execution.Ref{{SessionID: "session", TurnID: "one"}, {SessionID: "session", TurnID: "two"}}) {

v2-server/internal/agentruntime/retry_guard.go

@@ -0,0 +1,98 @@
+package agentruntime
+
+import (
+	"net/http"
+	"sync"
+
+	"github.com/wanglongan587/cloud/v2-server/internal/nodeservices"
+)
+
+// Agents re-send a model call that was cut off as if the network had failed: OpenCode up to
+// five times, Claude Code ten or more, each re-send streaming its partial answer again. A cut
+// at a model limit is not a network failure, so one turn allows only what the limit intends:
+// none after the total limit, one after an idle cut. Every later call of the turn is refused
+// before any upstream work, and the turn reports the limit as its failure.
+type retryGuard struct {
+	mu sync.Mutex
+	// idleCuts counts stalls (idle cuts of calls started after the previous cut) since the last
+	// call that started after a cut and completed normally.
+	idleCuts int
+	// calls numbers the turn's calls as they start; cutAfter is the number of the newest call
+	// started when the latest cut happened, so only a call started later can clear it.
+	calls, cutAfter int
+	// last is the limit that cut the turn's most recent call, cleared by a later completed call.
+	last string
+	// stopped is the limit that ended the turn; every later call is refused with it.
+	stopped string
+}
+
+var refusals = map[string]string{
+	nodeservices.CutMaxDuration: "模型请求超过时长上限被切断，这一轮已停止",
+	nodeservices.CutIdle:        "模型多次长时间没有输出，这一轮已停止",
+}
+
+func (g *retryGuard) provider(next nodeservices.Provider, protocol nodeservices.Protocol) nodeservices.Provider {
+	return guardedProvider{g, next, protocol}
+}
+
+// failure is the limit a turn ended at, or "" when its last call was not cut at a limit.
+func (g *retryGuard) failure() string {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if g.stopped != "" {
+		return g.stopped
+	}
+	return g.last
+}
+
+func (g *retryGuard) begin() (stopped string, call int) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.calls++
+	return g.stopped, g.calls
+}
+
+// finish records how call ended: completed is true only for a 2xx answer nothing cut, and limit
+// is the model limit that stopped it, before or after its header. Any other failure (a provider
+// rejection, an unreachable provider, a cancelled call) changes nothing: re-sending those is the
+// Agent's own policy.
+func (g *retryGuard) finish(call int, completed bool, limit string) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	switch {
+	case limit == nodeservices.CutMaxDuration:
+		g.last, g.stopped, g.cutAfter = limit, limit, g.calls
+	case limit == nodeservices.CutIdle:
+		// Calls that were already running at the previous cut stalled with it (a title request
+		// beside the turn's call): one stall, however many calls it cuts, earns one retry.
+		if call > g.cutAfter {
+			g.idleCuts++
+		}
+		g.last, g.cutAfter = limit, g.calls
+		if g.idleCuts > 1 {
+			g.stopped = limit
+		}
+	case completed:
+		// A call running alongside the cut one proves nothing about the retry.
+		if call > g.cutAfter {
+			g.idleCuts, g.last = 0, ""
+		}
+	}
+}
+
+type guardedProvider struct {
+	guard    *retryGuard
+	next     nodeservices.Provider
+	protocol nodeservices.Protocol
+}
+
+func (p guardedProvider) Complete(w http.ResponseWriter, r *http.Request, m nodeservices.Model) {
+	stopped, n := p.guard.begin()
+	if stopped != "" {
+		nodeservices.RefuseCall(w, p.protocol, stopped, refusals[stopped])
+		return
+	}
+	p.next.Complete(w, r, m)
+	completed, limit := nodeservices.Outcome(w)
+	p.guard.finish(n, completed, limit)
+}