diff --git a/internal/app/runtime_api.go b/internal/app/runtime_api.go index 3083b96f8..0901c8d43 100644 --- a/internal/app/runtime_api.go +++ b/internal/app/runtime_api.go @@ -6,6 +6,7 @@ import ( "github.com/uvwt/agentdock/internal/buildinfo" "github.com/uvwt/agentdock/internal/config" + "github.com/uvwt/agentdock/internal/observability" toolmcp "github.com/uvwt/agentdock/internal/tool/mcp" toolplugin "github.com/uvwt/agentdock/internal/tool/plugin" ) @@ -48,6 +49,16 @@ func (r *Runtime) RuntimeAnalytics() Result { } } +// RuntimeDiagnostics 只暴露最近调用的零 Payload 投影，供 Nexus 按需远程排障。 +// 本地 analytics 的进程指标与聚合统计不进入跨节点契约。 +func (r *Runtime) RuntimeDiagnostics() Result { + return Result{ + "ok": true, + "source": runtimeAPISource, + "recent_calls": observability.ProjectDiagnostics(r.observer.RecentCalls()), + } +} + func (r *Runtime) RuntimeSkills() (Result, error) { return r.skills.RuntimeSkills() } diff --git a/internal/app/runtime_observability_test.go b/internal/app/runtime_observability_test.go index f26fed9aa..f1a90ee40 100644 --- a/internal/app/runtime_observability_test.go +++ b/internal/app/runtime_observability_test.go @@ -135,3 +135,29 @@ func TestRuntimeToolLogCorrelatesRemoteTraceWithoutFakeChildSpan(t *testing.T) { t.Fatalf("tool log invented a local child span id: %s", body) } } + +func TestRuntimeDiagnosticsExposeOnlyRemoteSafeRecentCalls(t *testing.T) { + runtime := newRuntimeValidationTestRuntime(t) + ctx := observability.WithSource(context.Background(), observability.SourceNexus) + secret := "/Users/example/private/secret.txt" + if _, err := runtime.Call(ctx, "agentdock_context", map[string]any{"future_field": secret}); err == nil { + t.Fatal("expected validation error") + } + + encoded, err := json.Marshal(runtime.RuntimeDiagnostics()) + if err != nil { + t.Fatal(err) + } + body := strings.ToLower(string(encoded)) + for _, forbidden := range []string{ + strings.ToLower(secret), "span_id", "tool_stats", "process", "p50_duration_ms", + "arguments", "output", "command", "path", "error_message", "stack", + } { + if strings.Contains(body, forbidden) { + t.Fatalf("runtime diagnostics leaked %q: %s", forbidden, body) + } + } + if !strings.Contains(body, `"error_code":"invalid_argument"`) { + t.Fatalf("runtime diagnostics missing stable error code: %s", body) + } +} diff --git a/internal/httpx/runtime_api.go b/internal/httpx/runtime_api.go index 2b9edb53f..2fffc85f7 100644 --- a/internal/httpx/runtime_api.go +++ b/internal/httpx/runtime_api.go @@ -20,6 +20,7 @@ func registerRuntimeAPI(mux *http.ServeMux, runtime runtimeapi.Runtime, cfg conf h := runtimeAPIHandler(runtime, cfg, oauthStore) mux.HandleFunc("/internal/runtime/status", h) mux.HandleFunc("/internal/runtime/analytics", h) + mux.HandleFunc("/internal/runtime/diagnostics", h) mux.HandleFunc("/internal/runtime/capabilities", h) mux.HandleFunc("/internal/runtime/skills", h) mux.HandleFunc("/internal/runtime/skills/", h) diff --git a/internal/httpx/server_test.go b/internal/httpx/server_test.go index c4c96b4ff..6a2883cc3 100644 --- a/internal/httpx/server_test.go +++ b/internal/httpx/server_test.go @@ -340,6 +340,7 @@ func TestRuntimeAPIMethodContract(t *testing.T) { allow string }{ {method: http.MethodGet, path: "/internal/runtime/status", status: http.StatusOK}, + {method: http.MethodGet, path: "/internal/runtime/diagnostics", status: http.StatusOK}, {method: http.MethodPost, path: "/internal/runtime/capabilities", status: http.StatusOK}, {method: http.MethodPost, path: "/internal/runtime/status", status: http.StatusMethodNotAllowed, allow: "GET"}, {method: http.MethodPost, path: "/internal/runtime/tasks/tsk_1234567890abcdef", status: http.StatusMethodNotAllowed, allow: "GET, DELETE"}, diff --git a/internal/observability/diagnostics.go b/internal/observability/diagnostics.go new file mode 100644 index 000000000..762c1c80b --- /dev/null +++ b/internal/observability/diagnostics.go @@ -0,0 +1,59 @@ +package observability + +import ( + "strconv" + "time" +) + +// DiagnosticStage 是远程诊断允许跨进程暴露的阶段字段。 +// 它有意与 StageRecord 分离，避免未来给本地阶段模型增加字段时扩大远程契约。 +type DiagnosticStage struct { + Name string `json:"name"` + StartedOffsetMS float64 `json:"started_offset_ms"` + DurationMS float64 `json:"duration_ms"` + Success bool `json:"success"` +} + +// DiagnosticCall 是 Nexus 远程诊断消费的最小调用视图。 +// 不包含 SpanID、参数、结果、命令、路径、错误正文或其他 Payload。 +type DiagnosticCall struct { + ID string `json:"id"` + Tool string `json:"tool"` + Source string `json:"source"` + TraceID string `json:"trace_id,omitempty"` + StartedAt time.Time `json:"started_at"` + DurationMS float64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code,omitempty"` + ErrorCategory string `json:"error_category,omitempty"` + Stages []DiagnosticStage `json:"stages,omitempty"` +} + +// ProjectDiagnostics 把本地运行分析模型裁剪成稳定的远程诊断契约。 +func ProjectDiagnostics(records []ExecutionRecord) []DiagnosticCall { + out := make([]DiagnosticCall, 0, len(records)) + for _, record := range records { + stages := make([]DiagnosticStage, 0, len(record.Stages)) + for _, stage := range record.Stages { + stages = append(stages, DiagnosticStage{ + Name: string(stage.Name), + StartedOffsetMS: stage.StartedOffsetMS, + DurationMS: stage.DurationMS, + Success: stage.Success, + }) + } + out = append(out, DiagnosticCall{ + ID: strconv.FormatUint(record.ID, 10), + Tool: record.Tool, + Source: string(record.Source), + TraceID: record.TraceID, + StartedAt: record.StartedAt, + DurationMS: record.DurationMS, + Success: record.Success, + ErrorCode: record.ErrorCode, + ErrorCategory: record.ErrorCategory, + Stages: stages, + }) + } + return out +} diff --git a/internal/observability/diagnostics_test.go b/internal/observability/diagnostics_test.go new file mode 100644 index 000000000..02e3ee207 --- /dev/null +++ b/internal/observability/diagnostics_test.go @@ -0,0 +1,63 @@ +package observability + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +func TestRecentCallsAndDiagnosticsProjectionStayZeroPayload(t *testing.T) { + recorder := NewRecorder(4) + recorder.BeginTool() + recorder.EndTool( + "exec_command", + SourceNexus, + time.Unix(10, 0), + 25*time.Millisecond, + false, + "COMMAND_FAILED", + "runtime", + "4bf92f3577b34da6a3ce929d0e0e4736", + "00f067aa0ba902b7", + []StageRecord{{Name: StageCommandStart, StartedOffsetMS: 2, DurationMS: 3, Success: true}}, + ) + + projected := ProjectDiagnostics(recorder.RecentCalls()) + if len(projected) != 1 { + t.Fatalf("projected calls = %d", len(projected)) + } + call := projected[0] + if call.ID != "1" || call.Tool != "exec_command" || call.TraceID == "" || len(call.Stages) != 1 { + t.Fatalf("diagnostic call = %#v", call) + } + + encoded, err := json.Marshal(projected) + if err != nil { + t.Fatal(err) + } + body := strings.ToLower(string(encoded)) + for _, forbidden := range []string{ + "span_id", "arguments", "result", "output", "stdout", "stderr", + "command", "path", "error_message", "stack", "tool_stats", "process", + } { + if strings.Contains(body, `"`+forbidden+`"`) { + t.Fatalf("diagnostics JSON contains forbidden field %q: %s", forbidden, body) + } + } +} + +func TestRecentCallsReturnsIndependentStageCopy(t *testing.T) { + recorder := NewRecorder(2) + recorder.BeginTool() + recorder.EndTool( + "tool", SourceMCP, time.Now(), time.Millisecond, true, "", "", "", "", + []StageRecord{{Name: StageMCPRefresh, DurationMS: 1, Success: true}}, + ) + first := recorder.RecentCalls() + first[0].Stages[0].DurationMS = 999 + second := recorder.RecentCalls() + if second[0].Stages[0].DurationMS != 1 { + t.Fatalf("stored stage mutated through caller copy: %#v", second[0].Stages[0]) + } +} diff --git a/internal/observability/recorder.go b/internal/observability/recorder.go index 8127f0996..b7c50d1f5 100644 --- a/internal/observability/recorder.go +++ b/internal/observability/recorder.go @@ -150,6 +150,20 @@ func (r *Recorder) EndTool( } } +// RecentCalls 返回当前内存环里最近调用的独立副本，不计算百分位或进程指标。 +func (r *Recorder) RecentCalls() []ExecutionRecord { + if r == nil { + return []ExecutionRecord{} + } + r.mu.RLock() + recent := r.recentLocked() + r.mu.RUnlock() + for i := range recent { + recent[i].Stages = append([]StageRecord(nil), recent[i].Stages...) + } + return recent +} + func (r *Recorder) Snapshot() Snapshot { if r == nil { return Snapshot{} diff --git a/internal/runtimeapi/dispatch.go b/internal/runtimeapi/dispatch.go index b21e21c3a..e9a5b3aaa 100644 --- a/internal/runtimeapi/dispatch.go +++ b/internal/runtimeapi/dispatch.go @@ -56,6 +56,14 @@ func Dispatch(ctx context.Context, runtime Runtime, request Request) (map[string return map[string]any(runtime.RuntimeStatus()), nil case path == "/internal/runtime/analytics": return map[string]any(runtime.RuntimeAnalytics()), nil + case path == "/internal/runtime/diagnostics": + diagnostics, ok := runtime.(DiagnosticsRuntime) + if !ok { + return nil, &app.ToolError{ + Code: "DIAGNOSTICS_UNSUPPORTED", Message: "runtime does not support diagnostics", Category: "not_found", + } + } + return map[string]any(diagnostics.RuntimeDiagnostics()), nil case path == "/internal/runtime/capabilities": refresh := strings.EqualFold(request.queryValue("refresh"), "true") || method == http.MethodPost result, err := runtime.RuntimeCapabilities(ctx, refresh) diff --git a/internal/runtimeapi/dispatch_test.go b/internal/runtimeapi/dispatch_test.go index 4295bfbb6..36dd1a9d7 100644 --- a/internal/runtimeapi/dispatch_test.go +++ b/internal/runtimeapi/dispatch_test.go @@ -65,6 +65,12 @@ func (r *runtimeStub) RuntimeEvolve(context.Context, map[string]any) (app.Result return app.Result{}, nil } +type diagnosticsRuntimeStub struct{ runtimeStub } + +func (r *diagnosticsRuntimeStub) RuntimeDiagnostics() app.Result { + return app.Result{"recent_calls": []any{"call"}} +} + func TestMethodContract(t *testing.T) { tests := []struct { method string @@ -74,6 +80,7 @@ func TestMethodContract(t *testing.T) { }{ {"GET", "/internal/runtime/status", "GET", true}, {"GET", "/internal/runtime/analytics", "GET", true}, + {"GET", "/internal/runtime/diagnostics", "GET", true}, {"POST", "/internal/runtime/capabilities", "GET, POST", true}, {"DELETE", "/internal/runtime/tasks/task-1", "GET, DELETE", true}, {"POST", "/internal/runtime/tasks/task-1", "GET, DELETE", false}, @@ -104,6 +111,31 @@ func TestDispatchAnalytics(t *testing.T) { } } +func TestDispatchDiagnostics(t *testing.T) { + result, err := Dispatch(context.Background(), &diagnosticsRuntimeStub{}, Request{ + Method: "GET", + Path: "/internal/runtime/diagnostics", + }) + if err != nil { + t.Fatal(err) + } + calls, ok := result["recent_calls"].([]any) + if !ok || len(calls) != 1 || calls[0] != "call" { + t.Fatalf("diagnostics result = %#v", result) + } +} + +func TestDispatchDiagnosticsRejectsUnsupportedRuntime(t *testing.T) { + _, err := Dispatch(context.Background(), &runtimeStub{}, Request{ + Method: "GET", + Path: "/internal/runtime/diagnostics", + }) + var toolErr *app.ToolError + if !errors.As(err, &toolErr) || toolErr.Code != "DIAGNOSTICS_UNSUPPORTED" || toolErr.Category != "not_found" { + t.Fatalf("unsupported diagnostics error = %#v", err) + } +} + func TestDispatchParsesTaskQuery(t *testing.T) { runtime := &runtimeStub{} result, err := Dispatch(context.Background(), runtime, Request{ diff --git a/internal/runtimeapi/runtime.go b/internal/runtimeapi/runtime.go index c1e050cc4..db9493557 100644 --- a/internal/runtimeapi/runtime.go +++ b/internal/runtimeapi/runtime.go @@ -32,6 +32,12 @@ type Runtime interface { // MCPOAuthRuntime 是可选能力：只有支持 Remote MCP OAuth 的 Runtime 才实现。 // 保持它独立于 Runtime 主接口，避免 callback relay 把所有测试替身和只读调用方一起扩展。 +// DiagnosticsRuntime 是增量只读能力。保持它独立于 Runtime 主接口， +// 让旧实现或测试替身能明确返回“不支持”，而不是被迫实现空方法。 +type DiagnosticsRuntime interface { + RuntimeDiagnostics() app.Result +} + type MCPOAuthRuntime interface { RuntimeMCPOAuthCallback(context.Context, oauthclient.CallbackResult) error }