1package ports
2
3import (
4 "errors"
5 "sync"
6 "sync/atomic"
7)
8
9// RuntimeMarkSchema is the version of the process-local performance trace.
10const RuntimeMarkSchema uint16 = 1
11
12// RuntimeMarkKind identifies a process-local runtime boundary. It deliberately
13// contains no protocol payload, cells, or rendering policy.
14type RuntimeMarkKind string
15
16const (
17 RuntimeInputInjected RuntimeMarkKind = "input_injected"
18 RuntimeResizeRequested RuntimeMarkKind = "resize_requested"
19 RuntimeResizeCommitted RuntimeMarkKind = "resize_committed"
20 RuntimeCaptureStart RuntimeMarkKind = "capture_start"
21 RuntimeCaptureEnd RuntimeMarkKind = "capture_end"
22 RuntimeComposeStart RuntimeMarkKind = "compose_start"
23 RuntimeComposeEnd RuntimeMarkKind = "compose_end"
24 RuntimeDiffStart RuntimeMarkKind = "diff_start"
25 RuntimeDiffEnd RuntimeMarkKind = "diff_end"
26 RuntimeQueueEnqueued RuntimeMarkKind = "queue_enqueued"
27 RuntimeQueueDequeued RuntimeMarkKind = "queue_dequeued"
28 RuntimeACKBlockedStart RuntimeMarkKind = "ack_blocked_start"
29 RuntimeACKBlockedEnd RuntimeMarkKind = "ack_blocked_end"
30 RuntimeEmitStart RuntimeMarkKind = "emit_start"
31 RuntimeEmitEnd RuntimeMarkKind = "emit_end"
32 RuntimeAdapterSendStart RuntimeMarkKind = "adapter_send_start"
33 RuntimeAdapterSendEnd RuntimeMarkKind = "adapter_send_end"
34 RuntimeAdapterReceiveStart RuntimeMarkKind = "adapter_receive_start"
35 RuntimeAdapterReceiveEnd RuntimeMarkKind = "adapter_receive_end"
36 RuntimeTerminalFlushed RuntimeMarkKind = "terminal_flushed"
37 RuntimeScreenSnapshot RuntimeMarkKind = "screen_snapshot"
38 RuntimeScreenDelta RuntimeMarkKind = "screen_delta"
39 RuntimeScreenResetRequested RuntimeMarkKind = "screen_reset_requested"
40 RuntimeTransportDiagnostic RuntimeMarkKind = "transport_diagnostic"
41)
42
43// RuntimeMark is a process-local observation record. Tick is intentionally
44// left zero by producers: the concrete process observer owns timestamping.
45type RuntimeMark struct {
46 Schema uint16
47 ProcessID string
48 Component string
49 Scenario string
50 Run uint64
51 Sequence uint64
52 RequestID uint64
53 Epoch uint64
54 Kind RuntimeMarkKind
55 Tick int64
56 Bytes uint64
57 Fragments uint64
58 Retransmits uint64
59 Pending uint64
60 AckRTTNanos int64
61 Valid bool
62}
63
64// RuntimeObserver receives process-local marks. Implementations must not let
65// a producer provide a timestamp in another clock domain.
66type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }
67
68// SerializedRuntimeObserver is the only observer contract accepted by runtime
69// transport and use-case hot paths. ObserveRuntime must return without waiting
70// for sink I/O. Flush and Close wait until every accepted mark has reached the
71// concrete observer; Close is idempotent and rejects later marks.
72type SerializedRuntimeObserver interface {
73 RuntimeObserver
74 Flush()
75 Close()
76}
77
78// NewSerializedRuntimeObserver isolates producers from a potentially slow
79// observer with one ordered worker. capacity bounds queued marks. If the queue
80// fills, dropped marks are represented by one ordered diagnostic mark before
81// reporting resumes, so a trace can never silently claim completeness.
82func NewSerializedRuntimeObserver(observer RuntimeObserver, capacity int) SerializedRuntimeObserver {
83 if observer == nil {
84 return nil
85 }
86 if capacity < 1 {
87 capacity = 1
88 }
89 o := &serializedRuntimeObserver{observer: observer, capacity: capacity, done: make(chan struct{})}
90 o.cond = sync.NewCond(&o.mu)
91 go o.run()
92 return o
93}
94
95// EnsureSerializedRuntimeObserver returns observer's existing serialized
96// owner when it has one. Otherwise it creates an owned worker. Callers close
97// the result only when owned is true, preventing nested workers and duplicate
98// lifecycle ownership when application wiring already serialized the process
99// observer.
100func EnsureSerializedRuntimeObserver(observer RuntimeObserver, capacity int) (serialized SerializedRuntimeObserver, owned bool) {
101 if observer == nil {
102 return nil, false
103 }
104 if serialized, ok := observer.(SerializedRuntimeObserver); ok {
105 return serialized, false
106 }
107 return NewSerializedRuntimeObserver(observer, capacity), true
108}
109
110type serializedRuntimeObserver struct {
111 observer RuntimeObserver
112 capacity int
113
114 mu sync.Mutex
115 cond *sync.Cond
116 marks []RuntimeMark
117 dropped uint64
118 closed bool
119 active bool
120 done chan struct{}
121}
122
123func (o *serializedRuntimeObserver) ObserveRuntime(mark RuntimeMark) {
124 if o == nil {
125 return
126 }
127 o.mu.Lock()
128 defer o.mu.Unlock()
129 if o.closed {
130 return
131 }
132 if o.dropped != 0 {
133 // Reserve space for both the gap marker and the first post-gap mark.
134 // This preserves the producer order around a bounded-queue loss.
135 if len(o.marks)+2 > o.capacity {
136 o.dropped++
137 return
138 }
139 o.marks = append(o.marks, runtimeObserverGapMark(o.dropped), mark)
140 o.dropped = 0
141 o.cond.Signal()
142 return
143 }
144 if len(o.marks) == o.capacity {
145 o.dropped = 1
146 return
147 }
148 o.marks = append(o.marks, mark)
149 o.cond.Signal()
150}
151
152func runtimeObserverGapMark(dropped uint64) RuntimeMark {
153 return NewRuntimeMark("observer", RuntimeTransportDiagnostic, dropped, false)
154}
155
156func (o *serializedRuntimeObserver) Flush() {
157 if o == nil {
158 return
159 }
160 o.mu.Lock()
161 for len(o.marks) != 0 || o.active || o.dropped != 0 {
162 o.cond.Signal()
163 o.cond.Wait()
164 }
165 closed := o.closed
166 o.mu.Unlock()
167 if closed {
168 <-o.done
169 }
170}
171
172func (o *serializedRuntimeObserver) Close() {
173 if o == nil {
174 return
175 }
176 o.mu.Lock()
177 if !o.closed {
178 o.closed = true
179 o.cond.Broadcast()
180 }
181 o.mu.Unlock()
182 <-o.done
183}
184
185func (o *serializedRuntimeObserver) run() {
186 defer close(o.done)
187 for {
188 o.mu.Lock()
189 for len(o.marks) == 0 && o.dropped == 0 && !o.closed {
190 o.cond.Wait()
191 }
192 if len(o.marks) == 0 && o.dropped != 0 {
193 o.marks = append(o.marks, runtimeObserverGapMark(o.dropped))
194 o.dropped = 0
195 }
196 if len(o.marks) == 0 && o.closed {
197 o.cond.Broadcast()
198 o.mu.Unlock()
199 return
200 }
201 mark := o.marks[0]
202 o.marks = o.marks[1:]
203 o.active = true
204 o.mu.Unlock()
205
206 o.observer.ObserveRuntime(mark)
207
208 o.mu.Lock()
209 o.active = false
210 o.cond.Broadcast()
211 o.mu.Unlock()
212 }
213}
214
215// RuntimeMarkValid reports whether a mark has all correlation fields required
216// by the JSONL schema. Tick is checked separately by the timestamp owner.
217func RuntimeMarkValid(mark RuntimeMark) bool {
218 if mark.Schema != RuntimeMarkSchema || mark.Component == "" || mark.Scenario == "" || mark.Run == 0 || mark.Sequence == 0 || mark.RequestID == 0 || mark.Epoch == 0 {
219 return false
220 }
221 switch mark.Kind {
222 case RuntimeInputInjected, RuntimeResizeRequested, RuntimeResizeCommitted,
223 RuntimeCaptureStart, RuntimeCaptureEnd, RuntimeComposeStart, RuntimeComposeEnd,
224 RuntimeDiffStart, RuntimeDiffEnd, RuntimeQueueEnqueued, RuntimeQueueDequeued,
225 RuntimeACKBlockedStart, RuntimeACKBlockedEnd, RuntimeEmitStart, RuntimeEmitEnd,
226 RuntimeAdapterSendStart, RuntimeAdapterSendEnd, RuntimeAdapterReceiveStart,
227 RuntimeAdapterReceiveEnd, RuntimeTerminalFlushed, RuntimeScreenSnapshot,
228 RuntimeScreenDelta, RuntimeScreenResetRequested, RuntimeTransportDiagnostic:
229 return true
230 default:
231 return false
232 }
233}
234
235// RuntimeCorrelation identifies one attempted operation. It is deliberately
236// copied to both endpoints of a span; process ID remains owned by the sink.
237type RuntimeCorrelation struct {
238 Scenario string
239 Run, Sequence, RequestID, Epoch uint64
240}
241
242// RuntimeCorrelationInputs are supplied once by process orchestration. They
243// are deliberately data-only: consumer APIs continue to receive only their
244// one RuntimeObserver argument and never a clock or tick.
245type RuntimeCorrelationInputs struct {
246 Scenario string
247 Run uint64
248}
249
250// NewRuntimeCorrelationObserver binds every mark from one OS process to its
251// harness manifest scenario/run. Operation IDs remain producer-owned when
252// supplied; missing IDs are assigned from the process-wide allocator.
253func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorrelationInputs) (RuntimeObserver, error) {
254 if observer == nil || inputs.Scenario == "" || inputs.Run == 0 {
255 return nil, errors.New("invalid runtime correlation inputs")
256 }
257 return &runtimeCorrelationObserver{observer: observer, inputs: inputs}, nil
258}
259
260type runtimeCorrelationObserver struct {
261 observer RuntimeObserver
262 inputs RuntimeCorrelationInputs
263}
264
265func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {
266 if o == nil || o.observer == nil {
267 return
268 }
269 mark.Scenario, mark.Run = o.inputs.Scenario, o.inputs.Run
270 if mark.Sequence == 0 {
271 mark.Sequence = runtimeMarkSequence.Add(1)
272 }
273 if mark.RequestID == 0 {
274 mark.RequestID = mark.Sequence
275 }
276 if mark.Epoch == 0 {
277 mark.Epoch = mark.RequestID
278 }
279 o.observer.ObserveRuntime(mark)
280}
281
282var runtimeMarkSequence atomic.Uint64
283
284// NewRuntimeCorrelation makes a unique, process-local operation identity.
285// Components with a domain request/epoch should use those values directly via
286// NewRuntimeMarkWithCorrelation instead of inventing a second identity.
287func NewRuntimeCorrelation() RuntimeCorrelation {
288 sequence := runtimeMarkSequence.Add(1)
289 return RuntimeCorrelation{Scenario: "runtime", Run: 1, Sequence: sequence, RequestID: sequence, Epoch: sequence}
290}
291
292// NewRuntimeMarkWithCorrelation returns a timestamp-free mark carrying the
293// supplied operation identity. Start/end callers must reuse correlation.
294func NewRuntimeMarkWithCorrelation(component string, correlation RuntimeCorrelation, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
295 return RuntimeMark{Schema: RuntimeMarkSchema, Component: component, Scenario: correlation.Scenario, Run: correlation.Run, Sequence: correlation.Sequence, RequestID: correlation.RequestID, Epoch: correlation.Epoch, Kind: kind, Bytes: bytes, Valid: valid}
296}
297
298// NewRuntimeMark is retained for one-shot marks. Unlike the old helper it
299// never returns constant correlation IDs.
300func NewRuntimeMark(component string, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
301 return NewRuntimeMarkWithCorrelation(component, NewRuntimeCorrelation(), kind, bytes, valid)
302}