Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions internal/app/runtime_api.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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()
}
Expand Down
26 changes: 26 additions & 0 deletions internal/app/runtime_observability_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
1 change: 1 addition & 0 deletions internal/httpx/runtime_api.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
1 change: 1 addition & 0 deletions internal/httpx/server_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
Expand Down
59 changes: 59 additions & 0 deletions internal/observability/diagnostics.go
Original file line number Diff line number Diff line change
@@ -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
}
63 changes: 63 additions & 0 deletions internal/observability/diagnostics_test.go
Original file line number Diff line number Diff line change
@@ -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])
}
}
14 changes: 14 additions & 0 deletions internal/observability/recorder.go
Original file line number Diff line number Diff line change
Expand Up @@ -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{}
Expand Down
8 changes: 8 additions & 0 deletions internal/runtimeapi/dispatch.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
32 changes: 32 additions & 0 deletions internal/runtimeapi/dispatch_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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},
Expand Down Expand Up @@ -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{
Expand Down
6 changes: 6 additions & 0 deletions internal/runtimeapi/runtime.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
Loading