diff --git a/internal/runtime/trace/json_recorder.go b/internal/runtime/trace/json_recorder.go new file mode 100644 index 0000000..3a5ddc3 --- /dev/null +++ b/internal/runtime/trace/json_recorder.go @@ -0,0 +1,41 @@ +package trace + +import ( + "encoding/json" + "fmt" + "io" + "sync" +) + +// Writer persists a completed span. Implementations must be safe for concurrent +// use, as WriteSpan may be called from several goroutines. A Writer such as +// *JSONRecorder provides structured local logging for the M9 panel. +type Writer interface { + // WriteSpan persists a single completed span and returns any encoding or + // write error. + WriteSpan(rec SpanRecord) error +} + +// JSONRecorder writes each completed span as one line of JSON (JSONL) to an +// underlying io.Writer. It is safe for concurrent use. +type JSONRecorder struct { + mu sync.Mutex + enc *json.Encoder +} + +// NewJSONRecorder builds a JSONRecorder that appends one JSON object per span to +// out. +func NewJSONRecorder(out io.Writer) *JSONRecorder { + return &JSONRecorder{enc: json.NewEncoder(out)} +} + +// WriteSpan encodes rec as JSON followed by a newline. Concurrent calls are +// serialized so a single span object is never interleaved with another. +func (r *JSONRecorder) WriteSpan(rec SpanRecord) error { + r.mu.Lock() + defer r.mu.Unlock() + if err := r.enc.Encode(rec); err != nil { + return fmt.Errorf("trace: encode span %q: %w", rec.ID, err) + } + return nil +} diff --git a/internal/runtime/trace/span.go b/internal/runtime/trace/span.go new file mode 100644 index 0000000..abaa348 --- /dev/null +++ b/internal/runtime/trace/span.go @@ -0,0 +1,174 @@ +package trace + +import ( + "errors" + "sync" + "time" +) + +// ErrSpanFinished is returned by Span.Record and Span.SetProp when called on a +// span that has already been closed by Finish. A closed span is immutable. +var ErrSpanFinished = errors.New("trace: span already finished") + +// SpanKind classifies what a span represents so the M9 panel can group and color +// nodes by stage of the run. The values mirror the span kinds listed in +// DevDocs/design/04-运行时.md §9. +type SpanKind string + +const ( + // SpanKindLLM is a single LLM invocation. + SpanKindLLM SpanKind = "llm_call" + // SpanKindTool is a tool/function call. + SpanKindTool SpanKind = "tool_call" + // SpanKindOrchestration is an orchestration node, e.g. a plan step. + SpanKindOrchestration SpanKind = "orchestration" + // SpanKindStateTransition is a runtime state transition. + SpanKindStateTransition SpanKind = "state_transition" + // SpanKindSubagent is a spawned sub-agent. + SpanKindSubagent SpanKind = "subagent" + // SpanKindMemory is a memory read or write. + SpanKindMemory SpanKind = "memory" + // SpanKindGuardrail is a guardrail input/output check. + SpanKindGuardrail SpanKind = "guardrail" +) + +// SpanStatus is the terminal outcome of a span. +type SpanStatus string + +const ( + // SpanStatusOK marks a successfully completed span. + SpanStatusOK SpanStatus = "ok" + // SpanStatusError marks a failed span. + SpanStatusError SpanStatus = "error" +) + +// Event is a fine-grained record attached to a span, e.g. a token, retry or +// decision. Events enrich the chain-of-thought view. +type Event struct { + // Time is when the event happened. Record sets it to now when left zero. + Time time.Time `json:"time"` + // Type is a short label identifying the event kind. + Type string `json:"type"` + // Props carries event-specific structured data. + Props map[string]any `json:"props,omitempty"` +} + +// Span is a single traced unit of work, safe for concurrent use. +// +// A span is created by (*Tracer).StartSpan, which also returns a context that +// carries the span as the current one. Runtime work then records detail with +// Record and SetProp, and closes the span once with Finish. ID, ParentID, Name +// and Kind are fixed at creation; Status, EndedAt and Duration are set when +// Finish runs. Reading a span's fields is safe after Finish returns. +type Span struct { + mu sync.RWMutex + tracer *Tracer + + // ID is a span identifier unique across Tracers. + ID string `json:"id"` + // ParentID is the ID of the enclosing span, empty for a root span. + ParentID string `json:"parent_id,omitempty"` + // Name is a human-readable label for the work. + Name string `json:"name"` + // Kind classifies the span. + Kind SpanKind `json:"kind"` + // Status is the terminal outcome, valid after Finish. + Status SpanStatus `json:"status"` + // StartedAt is when the span began. + StartedAt time.Time `json:"started_at"` + // EndedAt is when Finish ran, zero until then. + EndedAt time.Time `json:"ended_at,omitempty"` + // Duration is EndedAt.Sub(StartedAt), zero until Finish. + Duration time.Duration `json:"duration_ns"` + // Props carries key/value metadata attached via SetProp. + Props map[string]any `json:"props,omitempty"` + // Events is the ordered list of events recorded via Record. + Events []Event `json:"events,omitempty"` +} + +// Record appends an event to the span. A zero Timestamp is stamped with now. +// Record returns ErrSpanFinished when the span has already been closed. +func (s *Span) Record(evt Event) error { + s.mu.Lock() + defer s.mu.Unlock() + if !s.EndedAt.IsZero() { + return ErrSpanFinished + } + if evt.Time.IsZero() { + evt.Time = time.Now() + } + s.Events = append(s.Events, evt) + return nil +} + +// SetProp attaches a key/value property to the span for structured output. +// SetProp returns ErrSpanFinished when the span has already been closed. +func (s *Span) SetProp(key string, value any) error { + s.mu.Lock() + defer s.mu.Unlock() + if !s.EndedAt.IsZero() { + return ErrSpanFinished + } + s.Props[key] = value + return nil +} + +// Finish closes the span with the given status, stamping EndedAt and Duration +// and handing the completed span to the owning Tracer's Writer. Finish is +// idempotent: calls after the first are ignored. +func (s *Span) Finish(status SpanStatus) { + s.mu.Lock() + if !s.EndedAt.IsZero() { + s.mu.Unlock() + return + } + s.Status = status + s.EndedAt = time.Now() + s.Duration = s.EndedAt.Sub(s.StartedAt) + rec := s.recordLocked() + s.mu.Unlock() + + if s.tracer != nil { + s.tracer.emit(rec) + } +} + +// recordLocked snapshots the span as a SpanRecord. The caller must hold s.mu. +func (s *Span) recordLocked() SpanRecord { + rec := SpanRecord{ + ID: s.ID, + ParentID: s.ParentID, + Name: s.Name, + Kind: s.Kind, + Status: s.Status, + StartedAt: s.StartedAt, + EndedAt: s.EndedAt, + Duration: s.Duration, + } + if len(s.Props) > 0 { + rec.Props = make(map[string]any, len(s.Props)) + for k, v := range s.Props { + rec.Props[k] = v + } + } + if len(s.Events) > 0 { + rec.Events = make([]Event, len(s.Events)) + copy(rec.Events, s.Events) + } + return rec +} + +// SpanRecord is an immutable, lock-free snapshot of a completed span, produced +// by Finish and passed to a Writer for persistence. It is safe to copy. +type SpanRecord struct { + ID string `json:"id"` + ParentID string `json:"parent_id,omitempty"` + Name string `json:"name"` + Kind SpanKind `json:"kind"` + Status SpanStatus `json:"status"` + StartedAt time.Time `json:"started_at"` + EndedAt time.Time `json:"ended_at,omitempty"` + Duration time.Duration `json:"duration_ns"` + Props map[string]any `json:"props,omitempty"` + Events []Event `json:"events,omitempty"` +} diff --git a/internal/runtime/trace/trace.go b/internal/runtime/trace/trace.go new file mode 100644 index 0000000..df317ea --- /dev/null +++ b/internal/runtime/trace/trace.go @@ -0,0 +1,113 @@ +// Package trace provides structured deep tracing for the Cyrene runtime. +// +// It records work as a tree of spans — one span per LLM call, tool call, +// orchestration node, state transition, sub-agent, memory access or guardrail — +// so the run can be rendered later as a flow graph and a chain-of-thought view +// (DevDocs/design/04-运行时.md §9). This package is a leaf: it depends only on +// the standard library and log/slog, and never imports the runtime packages +// (loop/agent) that consume it. +// +// Usage: create a Tracer over a Writer, then StartSpan a root and nested spans. +// StartSpan threads a parent/child chain through the context it returns, so +// runtime code only needs to pass that context along. A span is closed with +// Finish(status); finished spans are handed to the Writer (here JSONRecorder) +// for structured local logging. +package trace + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "log/slog" + "sync" + "time" +) + +// spanKey is the unexported context key that carries the current *Span. +type spanKey struct{} + +// Tracer creates, chains and completes spans. It is safe for concurrent use. +// +// A nil Writer is tolerated: spans are still created and chained, but nothing +// is emitted on Finish. Construct with New. +type Tracer struct { + mu sync.RWMutex + w Writer + seq uint64 + idPrefix string +} + +// New builds a Tracer that emits completed spans to w. Pass a Writer such as a +// *JSONRecorder to persist structured logs; pass nil to trace without emitting. +func New(w Writer) *Tracer { + return &Tracer{w: w, idPrefix: randPrefix()} +} + +// StartSpan begins a span named name of the given kind and returns it alongside +// a context that carries the span as the current one. If ctx already carries a +// span, the new span is chained as its child (its ParentID is set), enabling +// nested tracing; a nil ctx is treated as context.Background(). +func (t *Tracer) StartSpan(ctx context.Context, name string, kind SpanKind) (context.Context, *Span) { + if ctx == nil { + ctx = context.Background() + } + parent := SpanFromContext(ctx) + span := t.newSpan(name, kind, parent) + return context.WithValue(ctx, spanKey{}, span), span +} + +// SpanFromContext returns the span carried by ctx, or nil when ctx holds none +// (including a nil ctx). +func SpanFromContext(ctx context.Context) *Span { + if ctx == nil { + return nil + } + sp, _ := ctx.Value(spanKey{}).(*Span) + return sp +} + +// newSpan allocates a span, assigns it a unique ID, links ParentID and records +// its start time. +func (t *Tracer) newSpan(name string, kind SpanKind, parent *Span) *Span { + t.mu.Lock() + t.seq++ + id := fmt.Sprintf("%s-%d", t.idPrefix, t.seq) + t.mu.Unlock() + + sp := &Span{ + tracer: t, + ID: id, + Name: name, + Kind: kind, + StartedAt: time.Now(), + Props: make(map[string]any), + } + if parent != nil { + sp.ParentID = parent.ID + } + return sp +} + +// emit hands a finished span to the writer. Write failures are logged and never +// propagated, since Finish cannot return an error. It is a no-op for a nil +// writer. +func (t *Tracer) emit(rec SpanRecord) { + if t.w == nil { + return + } + if err := t.w.WriteSpan(rec); err != nil { + slog.Error("trace: write span failed", "span_id", rec.ID, "error", err) + } +} + +// randPrefix returns a short unique prefix used in span IDs so spans generated +// by different Tracers do not collide. It falls back to "tr" if the system +// entropy source is unavailable. +func randPrefix() string { + var b [8]byte + if _, err := rand.Read(b[:]); err != nil { + return "tr" + } + return hex.EncodeToString(b[:]) +} diff --git a/internal/runtime/trace/trace_test.go b/internal/runtime/trace/trace_test.go new file mode 100644 index 0000000..073139f --- /dev/null +++ b/internal/runtime/trace/trace_test.go @@ -0,0 +1,333 @@ +package trace + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" +) + +// decodeLines reads every JSON object emitted as JSONL into recs, failing the +// test if any line is malformed. +func decodeLines(t *testing.T, data []byte) []SpanRecord { + t.Helper() + sc := bufio.NewScanner(bytes.NewReader(data)) + var recs []SpanRecord + for sc.Scan() { + line := sc.Bytes() + if len(bytes.TrimSpace(line)) == 0 { + continue + } + var r SpanRecord + if err := json.Unmarshal(line, &r); err != nil { + t.Fatalf("decode span: %v", err) + } + recs = append(recs, r) + } + if err := sc.Err(); err != nil { + t.Fatalf("scan spans: %v", err) + } + return recs +} + +func TestSpan_Lifecycle(t *testing.T) { + t.Parallel() + var buf bytes.Buffer + tr := New(NewJSONRecorder(&buf)) + + ctx, sp := tr.StartSpan(context.Background(), "llm-call", SpanKindLLM) + if sp.ID == "" { + t.Fatal("span ID is empty") + } + if sp.Kind != SpanKindLLM || sp.Name != "llm-call" { + t.Fatalf("unexpected (name, kind) = (%q, %q)", sp.Name, sp.Kind) + } + if sp.StartedAt.IsZero() { + t.Fatalf("StartedAt is zero, want set at creation") + } + if sp.Status != "" || sp.Duration != 0 || !sp.EndedAt.IsZero() { + t.Fatalf("Status=%q Duration=%v EndedAt=%v, want zero before Finish", sp.Status, sp.Duration, sp.EndedAt) + } + if got := SpanFromContext(ctx); got != sp { + t.Fatalf("SpanFromContext = %p, want %p", got, sp) + } + + before := time.Now() + if err := sp.SetProp("model", "gpt-test"); err != nil { + t.Fatalf("SetProp: %v", err) + } + if err := sp.Record(Event{Type: "token", Props: map[string]any{"n": 1}}); err != nil { + t.Fatalf("Record: %v", err) + } + + sp.Finish(SpanStatusOK) + + if sp.Status != SpanStatusOK { + t.Fatalf("Status = %q, want ok", sp.Status) + } + if sp.EndedAt.Before(before) { + t.Fatalf("EndedAt %v is before snapshot %v", sp.EndedAt, before) + } + if sp.Duration <= 0 { + t.Fatalf("Duration = %v, want > 0", sp.Duration) + } + if len(sp.Events) != 1 || sp.Events[0].Type != "token" { + t.Fatalf("Events = %+v, want one 'token' event", sp.Events) + } + if sp.Props["model"] != "gpt-test" { + t.Fatalf("Props = %+v, want model key", sp.Props) + } + + recs := decodeLines(t, buf.Bytes()) + if len(recs) != 1 { + t.Fatalf("emitted %d records, want 1", len(recs)) + } +} + +func TestSpan_RecordStampsTime(t *testing.T) { + t.Parallel() + tr := New(nil) + _, sp := tr.StartSpan(context.Background(), "x", SpanKindTool) + + if err := sp.Record(Event{Type: "retry"}); err != nil { + t.Fatalf("Record: %v", err) + } + if sp.Events[0].Time.IsZero() { + t.Fatalf("event time was not stamped: %v", sp.Events[0].Time) + } +} + +func TestSpan_FinishTwiceIgnored(t *testing.T) { + t.Parallel() + var buf bytes.Buffer + tr := New(NewJSONRecorder(&buf)) + _, sp := tr.StartSpan(context.Background(), "x", SpanKindGuardrail) + + sp.Finish(SpanStatusError) + first := sp.Duration + sp.Finish(SpanStatusOK) + + if sp.Status != SpanStatusError { + t.Fatalf("Status = %q, want error (second Finish ignored)", sp.Status) + } + if sp.Duration != first { + t.Fatalf("Duration changed from %v to %v on second Finish", first, sp.Duration) + } + if recs := decodeLines(t, buf.Bytes()); len(recs) != 1 { + t.Fatalf("emitted %d records, want 1", len(recs)) + } +} + +func TestSpan_MutationAfterFinish(t *testing.T) { + t.Parallel() + tr := New(nil) + _, sp := tr.StartSpan(context.Background(), "x", SpanKindMemory) + sp.Finish(SpanStatusOK) + + if err := sp.Record(Event{Type: "late"}); err != ErrSpanFinished { + t.Fatalf("Record after Finish = %v, want ErrSpanFinished", err) + } + if err := sp.SetProp("late", true); err != ErrSpanFinished { + t.Fatalf("SetProp after Finish = %v, want ErrSpanFinished", err) + } + if len(sp.Events) != 0 || len(sp.Props) != 0 { + t.Fatalf("closed span mutated: events=%d props=%d", len(sp.Events), len(sp.Props)) + } +} + +func TestTracer_ParentChildChain(t *testing.T) { + t.Parallel() + tr := New(nil) + + ctx, root := tr.StartSpan(context.Background(), "run", SpanKindOrchestration) + ctx, child := tr.StartSpan(ctx, "llm", SpanKindLLM) + + if child.ParentID != root.ID { + t.Fatalf("child.ParentID = %q, want root %q", child.ParentID, root.ID) + } + if root.ParentID != "" { + t.Fatalf("root.ParentID = %q, want empty", root.ParentID) + } + if got := SpanFromContext(ctx); got != child { + t.Fatalf("SpanFromContext = %p, want child %p", got, child) + } + + // A sibling span is chained to the same parent via the parent's context. + _, sibling := tr.StartSpan(context.Background(), "tool", SpanKindTool) + if sibling.ParentID != "" { + t.Fatalf("sibling.ParentID = %q, want empty (no parent in fresh ctx)", sibling.ParentID) + } +} + +func TestTracer_TwoLevels(t *testing.T) { + t.Parallel() + tr := New(nil) + ctx, root := tr.StartSpan(context.Background(), "run", SpanKindOrchestration) + ctx, mid := tr.StartSpan(ctx, "plan", SpanKindOrchestration) + ctx, leaf := tr.StartSpan(ctx, "llm", SpanKindLLM) + + if leaf.ParentID != mid.ID { + t.Fatalf("leaf.ParentID = %q, want mid %q", leaf.ParentID, mid.ID) + } + if mid.ParentID != root.ID { + t.Fatalf("mid.ParentID = %q, want root %q", mid.ParentID, root.ID) + } +} + +func TestTracer_NilAndEmptyContext(t *testing.T) { + t.Parallel() + tr := New(nil) + + if got := SpanFromContext(nil); got != nil { + t.Fatalf("SpanFromContext(nil) = %p, want nil", got) + } + if got := SpanFromContext(context.Background()); got != nil { + t.Fatalf("SpanFromContext(bg) = %p, want nil", got) + } + + if _, sp := tr.StartSpan(nil, "x", SpanKindStateTransition); sp == nil { + t.Fatal("StartSpan(nil ctx) returned nil span") + } +} + +func TestTracer_UniqueIDs(t *testing.T) { + t.Parallel() + tr := New(nil) + ids := make(map[string]bool) + for i := 0; i < 100; i++ { + _, sp := tr.StartSpan(context.Background(), "x", SpanKindTool) + if sp.ID == "" { + t.Fatal("empty span ID") + } + if ids[sp.ID] { + t.Fatalf("duplicate span ID %q", sp.ID) + } + ids[sp.ID] = true + } +} + +func TestJSONRecorder_Output(t *testing.T) { + t.Parallel() + var buf bytes.Buffer + tr := New(NewJSONRecorder(&buf)) + + ctx, root := tr.StartSpan(context.Background(), "run", SpanKindOrchestration) + _, child := tr.StartSpan(ctx, "llm", SpanKindLLM) + _ = child.SetProp("model", "gpt-test") + _ = child.Record(Event{Type: "prompt", Props: map[string]any{"tokens": 12}}) + child.Finish(SpanStatusOK) + root.Finish(SpanStatusOK) + + recs := decodeLines(t, buf.Bytes()) + if len(recs) != 2 { + t.Fatalf("emitted %d records, want 2: %+v", len(recs), recs) + } + + var childRec, rootRec SpanRecord + for _, r := range recs { + if r.Kind == SpanKindLLM { + childRec = r + } else { + rootRec = r + } + } + if childRec.ParentID != rootRec.ID { + t.Fatalf("childRec.ParentID = %q, want root %q", childRec.ParentID, rootRec.ID) + } + if childRec.Status != SpanStatusOK { + t.Fatalf("childRec.Status = %q, want ok", childRec.Status) + } + if childRec.Duration <= 0 { + t.Fatalf("childRec.Duration = %v, want > 0", childRec.Duration) + } + if childRec.Props["model"] != "gpt-test" { + t.Fatalf("childRec.Props = %+v, want model key", childRec.Props) + } + if len(childRec.Events) != 1 || childRec.Events[0].Type != "prompt" { + t.Fatalf("childRec.Events = %+v, want one 'prompt' event", childRec.Events) + } + + // Ensure lines are well-formed JSON objects (not interleaved), by re-decoding. + for _, r := range recs { + if r.ID == "" { + t.Fatal("record with empty ID was emitted") + } + } +} + +func TestTracer_Concurrent(t *testing.T) { + t.Parallel() + var buf bytes.Buffer + tr := New(NewJSONRecorder(&buf)) + + const workers = 64 + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ctx, parent := tr.StartSpan(context.Background(), "run", SpanKindOrchestration) + for j := 0; j < 5; j++ { + _, child := tr.StartSpan(ctx, "llm", SpanKindLLM) + if child.ParentID != parent.ID { + t.Errorf("child.ParentID = %q, want parent %q", child.ParentID, parent.ID) + return + } + _ = child.SetProp("worker", "w") + _ = child.Record(Event{Type: "done"}) + child.Finish(SpanStatusOK) + } + parent.Finish(SpanStatusOK) + }() + } + wg.Wait() + + recs := decodeLines(t, buf.Bytes()) + // Each worker emits one parent + 5 children. + if len(recs) != workers*6 { + t.Fatalf("emitted %d records, want %d", len(recs), workers*6) + } + + // No two records may share an ID. + seen := make(map[string]bool, len(recs)) + for _, r := range recs { + if seen[r.ID] { + t.Fatalf("duplicate ID %q emitted concurrently", r.ID) + } + seen[r.ID] = true + } +} + +func TestTracer_NilWriterSpansCreateOnly(t *testing.T) { + t.Parallel() + tr := New(nil) + _, sp := tr.StartSpan(context.Background(), "x", SpanKindSubagent) + sp.Finish(SpanStatusOK) + if sp.Status != SpanStatusOK { + t.Fatalf("Status = %q, want ok", sp.Status) + } +} + +func TestSpanKind_Values(t *testing.T) { + t.Parallel() + want := []SpanKind{ + SpanKindLLM, SpanKindTool, SpanKindOrchestration, SpanKindStateTransition, + SpanKindSubagent, SpanKindMemory, SpanKindGuardrail, + } + got := []string{ + "llm_call", "tool_call", "orchestration", "state_transition", + "subagent", "memory", "guardrail", + } + for i := range want { + if string(want[i]) != got[i] { + t.Fatalf("SpanKind[%d] = %q, want %q", i, want[i], got[i]) + } + } + if !strings.Contains(string(want[0]), "llm_call") { + t.Fatalf("kind string mismatch: %s", want[0]) + } +}