spb/forge-studio Public

1// Author: Simon-Pierre Boucher — contact@spboucher.ai2//3// Per-run time series. Append-only (raw data is never thrown away); the UI4// asks for downsampled snapshots sized to the chart's pixel width. An actor5// so live ingest and UI reads never race.6import Foundation78actor MetricsStore {9    private(set) var points: [MetricPoint] = []10    private var parser = LogParser()11    private var csvOffset: UInt64 = 01213    struct Snapshot: Sendable {14        var train: [Downsampler.XY]15        var trainEMA: [Downsampler.XY]16        var val: [Downsampler.XY]17        var lr: [Downsampler.XY]18        var tokensPerSec: [Downsampler.XY]19        var gradNorm: [Downsampler.XY]20        var lastPoint: MetricPoint?21        var bestVal: (step: Int, loss: Double)?22        var count: Int23        var lastGrowth: Date?  // watchdog: when new rows last arrived24    }2526    private var lastGrowth: Date?2728    func append(_ p: MetricPoint) { points.append(p) }2930    func reset() {31        points.removeAll()32        parser = LogParser()33        csvOffset = 034    }3536    /// Incremental tail of log.csv: reads only bytes past the last offset,37    /// so polling during a live run costs O(new lines).38    func ingestCSV(at url: URL) {39        guard let fh = try? FileHandle(forReadingFrom: url) else { return }40        defer { try? fh.close() }41        try? fh.seek(toOffset: csvOffset)42        guard let data = try? fh.readToEnd(), !data.isEmpty else { return }43        // Only consume complete lines; leave a partial tail for next poll.44        var consumable = data45        if let lastNL = data.lastIndex(of: 0x0A) {46            consumable = data[data.startIndex...lastNL]47        } else {48            return49        }50        csvOffset += UInt64(consumable.count)51        guard let text = String(data: consumable, encoding: .utf8) else { return }52        var grew = false53        for line in text.split(separator: "\n") {54            if let p = parser.parseCSVLine(String(line)) {55                points.append(p)56                grew = true57            }58        }59        if grew { lastGrowth = .now }60    }6162    func snapshot(maxPoints: Int, smoothing: Double) -> Snapshot {63        func series(_ f: (MetricPoint) -> Double?) -> [Downsampler.XY] {64            points.compactMap { p in f(p).map { .init(x: Double(p.step), y: $0) } }65        }66        let train = series { $0.trainLoss }67        let emaValues = Smoothing.ema(train.map(\.y), smoothing: smoothing)68        let ema = zip(train, emaValues).map { Downsampler.XY(x: $0.x, y: $1) }69        let val = series { $0.valLoss }70        var best: (Int, Double)?71        for p in points {72            if let v = p.valLoss, v.isFinite, best == nil || v < best!.1 {73                best = (p.step, v)74            }75        }76        func ds(_ s: [Downsampler.XY]) -> [Downsampler.XY] {77            Downsampler.lttb(s, threshold: maxPoints)78        }79        return Snapshot(80            train: ds(train), trainEMA: ds(ema), val: val, // val is sparse: keep raw81            lr: ds(series { $0.lr }),82            tokensPerSec: ds(series { $0.tokensPerSec }),83            gradNorm: ds(series { $0.gradNorm }),84            lastPoint: points.last,85            bestVal: best,86            count: points.count,87            lastGrowth: lastGrowth)88    }89}90