module Observe.Stream import Data.UTF8 import Model.Word32 import Std.Byte import Std.Flag import Data.JSON import Data.JSONAccess import Observe.Value import Observe.Interchange import Std.List import Std.Natural -- AOS-JSON/1 STREAMS (docs/standards/AOS-1.0-draft-1.md §24.1, §10): a -- stream's verdict -- complete, incomplete or invalid -- and the reason. -- -- invalid a line that is not a valid record; a first record that is -- not a header; a record after the footer; a footer whose -- count is not the number of lines before it; an event whose -- sequence does not increase within its producer epoch -- incomplete bytes after the last LF (a torn write: AOS-REC-001's valid -- prefix with an incomplete tail); no footer (AOS-JSON-002); -- a footer that says incomplete; sequences missing beyond -- the drops the producer's coverage record declares -- (AOS-COV-004 .. 005) -- complete none of these: complete for what the stream declares, not -- a claim that every physical event was observed (AOS-COV-001) -- -- An incomplete stream is not an invalid one: its records remain -- individually valid evidence (AOS-OBS-004). Status and problem numbers are -- a contract with the bootstrap reader (tool/Aos.hs `StreamStatus`, -- `StreamProblem`); never renumber one. -- -- status 0 complete, 1 incomplete, 2 invalid -- problem 0 none, 1 empty, 2 invalid-line, 3 header-missing, -- 4 after-footer, 5 footer-count, 6 sequence-order, 7 no-footer, -- 8 footer-incomplete, 9 truncated-tail, 10 unaccounted-gap, -- 11 coverage-cumulative -- --------------------------------------------------------------------------- -- Lines family AosStreamVerdict : Type 0 constructor AosStreamVerdictValue field unrestricted aosVerdictStatus : Nat field unrestricted aosVerdictProblem : Nat field unrestricted aosVerdictLine : Nat field unrestricted aosVerdictLineCode : Nat field unrestricted aosVerdictLines : Nat field unrestricted aosVerdictEvents : Nat field unrestricted aosVerdictUnknownRecords : Nat field unrestricted aosVerdictTruncatedBytes : Nat field unrestricted aosVerdictGaps : Nat field unrestricted aosVerdictDeclaredDrops : Nat end-family -- one producer epoch's sequence state: the next sequence not yet seen (the -- last seen plus one; 0 before any), the events seen, the drops declared family AosProducer : Type 0 constructor AosProducerValue field unrestricted aosProducerKey : Bytes field unrestricted aosProducerNext : Nat field unrestricted aosProducerEvents : Nat field unrestricted aosProducerDrops : Nat field unrestricted aosProducerCoveredNext : Nat field unrestricted aosProducerExcluded : Nat field unrestricted aosProducerCoverageMode : Nat end-family family AosWalk : Type 0 constructor AosWalkValue field unrestricted aosWalkProblem : Nat field unrestricted aosWalkProblemLine : Nat field unrestricted aosWalkLineCode : Nat field unrestricted aosWalkProducers : (family StdList (family AosProducer)) field unrestricted aosWalkFooter : Nat field unrestricted aosWalkFooterComplete : Nat field unrestricted aosWalkEvents : Nat field unrestricted aosWalkUnknown : Nat end-family family AosFound : Type 0 constructor AosFoundNone constructor AosFoundSome field unrestricted aosFoundProducer : (family AosProducer) end-family def aosBytesCons = (lambda unrestricted head : Bytes . (lambda unrestricted rest : (family StdList Bytes) . (constructor StdList StdListCons Bytes head rest))) def aosListHead = (lambda unrestricted list : (family StdList Bytes) . (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . Bytes) list (branch StdListEmpty . b"") (branch StdListCons head tail induction . head))) def aosListTail = (lambda unrestricted list : (family StdList Bytes) . (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . (family StdList Bytes)) list (branch StdListEmpty . (constructor StdList StdListEmpty Bytes)) (branch StdListCons head tail induction . tail))) -- The stream split at every LF: n LFs give n + 1 segments, the last the -- bytes after the final LF (empty for a stream that ends with one). def aosSegments = (lambda unrestricted stream : Bytes . (bytes-eliminate (lambda unrestricted current : Bytes . (family StdList Bytes)) (aosBytesCons b"" (constructor StdList StdListEmpty Bytes)) (lambda unrestricted head : Byte . (lambda unrestricted tail : Bytes . (lambda unrestricted induction : (family StdList Bytes) . (nat-eliminate (lambda unrestricted current : Nat . (family StdList Bytes)) (aosBytesCons (bytes-cons head (aosListHead induction)) (aosListTail induction)) (lambda unrestricted predecessor : Nat . (lambda unrestricted inner : (family StdList Bytes) . (aosBytesCons b"" induction))) (byte-equal head (byte 10)))))) stream)) def aosListLength = (lambda unrestricted list : (family StdList Bytes) . (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . Nat) list (branch StdListEmpty . 0) (branch StdListCons head tail induction . (succ induction)))) def aosListLast = (lambda unrestricted list : (family StdList Bytes) . (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . Bytes) list (branch StdListEmpty . b"") (branch StdListCons head tail induction . (aosChoose Bytes (naturalIsZero (aosListLength tail)) head induction)))) -- --------------------------------------------------------------------------- -- Producers -- the producer epoch an event or a coverage record belongs to def aosProducerKeyOf = (lambda unrestricted identity : (family JSON) . (bytes-append (aosMemberBytes b"attempt_id" identity) (bytes-append b"/" (bytes-append (aosMemberBytes b"producer_id" identity) (bytes-append b"/" (aosMemberBytes b"producer_epoch" identity)))))) def aosKeyOf = (lambda unrestricted producer : (family AosProducer) . (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . Bytes) producer (branch AosProducerValue key next events drops covered excluded mode . key))) def aosFind = (lambda unrestricted key : Bytes . (lambda unrestricted producers : (family StdList (family AosProducer)) . (eliminate StdList (lambda unrestricted current : (family StdList (family AosProducer)) . (family AosFound)) producers (branch StdListEmpty . (constructor AosFound AosFoundNone)) (branch StdListCons head tail induction . (aosChoose (family AosFound) (bytes-equal (aosKeyOf head) key) (constructor AosFound AosFoundSome head) induction))))) -- the list with the producer of `key` replaced by `update` of what was there -- (appended when absent) def aosUpsert = (lambda unrestricted key : Bytes . (lambda unrestricted update : (pi unrestricted found : (family AosFound) . (family AosProducer)) . (lambda unrestricted producers : (family StdList (family AosProducer)) . (eliminate StdList (lambda unrestricted current : (family StdList (family AosProducer)) . (family StdList (family AosProducer))) producers (branch StdListEmpty . (constructor StdList StdListCons (family AosProducer) (update (constructor AosFound AosFoundNone)) (constructor StdList StdListEmpty (family AosProducer)))) (branch StdListCons head tail induction . (aosChoose (family StdList (family AosProducer)) (bytes-equal (aosKeyOf head) key) (constructor StdList StdListCons (family AosProducer) (update (constructor AosFound AosFoundSome head)) tail) (constructor StdList StdListCons (family AosProducer) head induction))))))) -- --------------------------------------------------------------------------- -- The walk def aosWalkInitial : (family AosWalk) = (constructor AosWalk AosWalkValue 0 0 0 (constructor StdList StdListEmpty (family AosProducer)) 0 0 0 0) -- the walk with a problem found at line n (it stops the walk) def aosWalkFail = (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted problem : Nat . (lambda unrestricted n : Nat . (lambda unrestricted code : Nat . (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . (family AosWalk)) walk (branch AosWalkValue oldProblem oldLine oldCode producers footer complete events unknown . (constructor AosWalk AosWalkValue problem n code producers footer complete events unknown))))))) def aosDecimalOf = (lambda unrestricted name : Bytes . (lambda unrestricted value : (family JSON) . (aosDigitsValue (aosMemberBytes name value)))) -- an event: its sequence must exceed every earlier one of its producer epoch def aosWalkEvent = (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . (lambda unrestricted record : (family JSON) . (let unrestricted identity = (aosMemberValue b"event_id" record) in (let unrestricted key = (aosProducerKeyOf identity) in (let unrestricted sequence = (aosDecimalOf b"sequence" identity) in (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . (family AosWalk)) walk (branch AosWalkValue problem line code producers footer complete events unknown . (let unrestricted outOfOrder = (eliminate AosFound (lambda unrestricted current : (family AosFound) . Nat) (aosFind key producers) (branch AosFoundNone . 0) (branch AosFoundSome producer . (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . Nat) producer (branch AosProducerValue k next seen drops covered excluded mode . (naturalOr (naturalLess sequence next) (naturalLess sequence covered)))))) in (aosChoose (family AosWalk) outOfOrder (aosWalkFail walk 6 n 0) (constructor AosWalk AosWalkValue problem line code (aosUpsert key (lambda unrestricted found : (family AosFound) . (eliminate AosFound (lambda unrestricted current : (family AosFound) . (family AosProducer)) found (branch AosFoundNone . (constructor AosProducer AosProducerValue key (succ sequence) 1 0 0 0 0)) (branch AosFoundSome producer . (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . (family AosProducer)) producer (branch AosProducerValue k next seen drops covered excluded mode . (constructor AosProducer AosProducerValue key (succ sequence) (succ seen) drops covered excluded mode)))))) producers) footer complete (succ events) unknown))))))))))) -- Legacy coverage adds interval counts. Critical cumulative coverage is an -- exact prefix snapshot: repeated checkpoints replace totals, never add them. -- One selected sequence is either accepted or dropped; policy exclusions -- consume no sequence. Shared by cumulative coverage and live observer views. def aosSequenceAccounting = (lambda unrestricted next : Nat . (lambda unrestricted accepted : Nat . (lambda unrestricted dropped : Nat . (naturalEqual next (naturalAdd accepted dropped))))) -- Accepted observations must equal the actual seen population; next includes -- trailing selected losses. A producer cannot mix the two coverage meanings. def aosCoverageConsistent = (lambda unrestricted producer : (family AosProducer) . (lambda unrestricted record : (family JSON) . (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . Nat) producer (branch AosProducerValue key next seen drops covered excluded mode . (aosChoose Nat (bytes-equal (aosMemberBytes b"coverage_mode" record) b"cumulative") (naturalAnd (naturalIsZero (naturalEqual mode 1)) (naturalAnd (naturalLessOrEqual next (aosDecimalOf b"next_sequence" record)) (naturalAnd (naturalLessOrEqual covered (aosDecimalOf b"next_sequence" record)) (naturalAnd (naturalLessOrEqual drops (aosDecimalOf b"records_dropped" record)) (naturalAnd (naturalLessOrEqual excluded (aosDecimalOf b"records_excluded" record)) (naturalAnd (naturalEqual seen (aosDecimalOf b"records_accepted" record)) (aosSequenceAccounting (aosDecimalOf b"next_sequence" record) seen (aosDecimalOf b"records_dropped" record)))))))) (naturalIsZero (naturalEqual mode 2))))))) def aosCoverageUpdate = (lambda unrestricted producer : (family AosProducer) . (lambda unrestricted record : (family JSON) . (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . (family AosProducer)) producer (branch AosProducerValue key next seen drops covered excluded mode . (aosChoose (family AosProducer) (bytes-equal (aosMemberBytes b"coverage_mode" record) b"cumulative") (constructor AosProducer AosProducerValue key next seen (aosDecimalOf b"records_dropped" record) (aosDecimalOf b"next_sequence" record) (aosDecimalOf b"records_excluded" record) 2) (constructor AosProducer AosProducerValue key next seen (naturalAdd drops (aosDecimalOf b"records_dropped" record)) covered excluded 1)))))) def aosCoveragePrior = (lambda unrestricted key : Bytes . (lambda unrestricted found : (family AosFound) . (eliminate AosFound (lambda unrestricted current : (family AosFound) . (family AosProducer)) found (branch AosFoundNone . (constructor AosProducer AosProducerValue key 0 0 0 0 0 0)) (branch AosFoundSome producer . producer)))) def aosWalkCoverage = (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . (lambda unrestricted record : (family JSON) . (let unrestricted key = (aosProducerKeyOf record) in (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . (family AosWalk)) walk (branch AosWalkValue problem line code producers footer complete events unknown . (aosChoose (family AosWalk) (aosCoverageConsistent (aosCoveragePrior key (aosFind key producers)) record) (constructor AosWalk AosWalkValue problem line code (aosUpsert key (lambda unrestricted found : (family AosFound) . (aosCoverageUpdate (aosCoveragePrior key found) record)) producers) footer complete events unknown) (aosWalkFail walk 11 n 0)))))))) -- the footer: it counts the lines before it def aosWalkFooter = (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . (lambda unrestricted record : (family JSON) . (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . (family AosWalk)) walk (branch AosWalkValue problem line code producers footer complete events unknown . (aosChoose (family AosWalk) (naturalEqual (aosDecimalOf b"records" record) (naturalSaturatingSubtract n 1)) (constructor AosWalk AosWalkValue problem line code producers 1 (aosBooleanFlag (aosMemberValue b"complete" record)) events unknown) (aosWalkFail walk 5 n 0))))))) def aosWalkUnknownRecord = (lambda unrestricted walk : (family AosWalk) . (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . (family AosWalk)) walk (branch AosWalkValue problem line code producers footer complete events unknown . (constructor AosWalk AosWalkValue problem line code producers footer complete events (succ unknown))))) def aosWalkProblemOf = (lambda unrestricted walk : (family AosWalk) . (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . Nat) walk (branch AosWalkValue problem line code producers footer complete events unknown . problem))) def aosWalkFooterOf = (lambda unrestricted walk : (family AosWalk) . (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . Nat) walk (branch AosWalkValue problem line code producers footer complete events unknown . footer))) -- a valid record at line n def aosWalkRecord = (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . (lambda unrestricted record : (family JSON) . (let unrestricted kind = (aosMemberBytes b"type" record) in (aosChoose (family AosWalk) (naturalEqual n 1) (aosChoose (family AosWalk) (bytes-equal kind b"header") walk (aosWalkFail walk 3 1 0)) (aosChoose (family AosWalk) (bytes-equal kind b"event") (aosWalkEvent walk n record) (aosChoose (family AosWalk) (bytes-equal kind b"coverage") (aosWalkCoverage walk n record) (aosChoose (family AosWalk) (bytes-equal kind b"footer") (aosWalkFooter walk n record) (aosChoose (family AosWalk) (naturalOr (bytes-equal kind b"header") (naturalOr (bytes-equal kind b"descriptor") (bytes-equal kind b"artifact"))) walk (aosWalkUnknownRecord walk)))))))))) -- line n of the stream: stops at the first problem def aosWalkLine = (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . (lambda unrestricted line : Bytes . (aosChoose (family AosWalk) (aosWalkProblemOf walk) walk (eliminate AosLine (lambda unrestricted current : (family AosLine) . (family AosWalk)) (aosClassifyLine line) (branch AosLineInvalid code . (aosWalkFail walk 2 n code)) (branch AosLineValid record . (aosChoose (family AosWalk) (aosWalkFooterOf walk) (aosWalkFail walk 4 n 0) (aosWalkRecord walk n record)))))))) -- the complete lines (every segment but the last), walked in order from -- line 1: a left fold, the walk threaded through the right fold's functions def aosWalkLinesFrom = (lambda unrestricted initial : (family AosWalk) . (lambda unrestricted firstLine : Nat . (lambda unrestricted segments : (family StdList Bytes) . (app (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . (pi unrestricted walk : (family AosWalk) . (pi unrestricted n : Nat . (family AosWalk)))) segments (branch StdListEmpty . (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . walk))) (branch StdListCons line rest induction . (lambda unrestricted walk : (family AosWalk) . (lambda unrestricted n : Nat . -- the last segment is not a line: it is the tail after the last LF (aosChoose (family AosWalk) (naturalIsZero (aosListLength rest)) walk (induction (aosWalkLine walk n line) (succ n))))))) initial firstLine)))) def aosWalkLines = (lambda unrestricted segments : (family StdList Bytes) . (aosWalkLinesFrom aosWalkInitial 1 segments)) -- --------------------------------------------------------------------------- -- The verdict def aosSumGaps = (lambda unrestricted producers : (family StdList (family AosProducer)) . (lambda unrestricted withDrops : Nat . (eliminate StdList (lambda unrestricted current : (family StdList (family AosProducer)) . Nat) producers (branch StdListEmpty . 0) (branch StdListCons head tail induction . (naturalAdd induction (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . Nat) head (branch AosProducerValue key next seen drops covered excluded mode . (naturalSaturatingSubtract (naturalSaturatingSubtract (aosChoose Nat (naturalLess next covered) covered next) seen) (naturalSelect withDrops drops 0))))))))) def aosSumDrops = (lambda unrestricted producers : (family StdList (family AosProducer)) . (eliminate StdList (lambda unrestricted current : (family StdList (family AosProducer)) . Nat) producers (branch StdListEmpty . 0) (branch StdListCons head tail induction . (naturalAdd induction (eliminate AosProducer (lambda unrestricted current : (family AosProducer) . Nat) head (branch AosProducerValue key next seen drops covered excluded mode . drops)))))) -- Finalization is separate from feeding so segment boundaries cannot reset -- sequence state, footer count, or the first diagnostic. Tail is its original -- byte extent, including a deliberately opaque oversized unfinished frame. def aosFinishStreamScan = (lambda unrestricted lines : Nat . (lambda unrestricted tailBytes : Nat . (lambda unrestricted walk : (family AosWalk) . (eliminate AosWalk (lambda unrestricted current : (family AosWalk) . (family AosStreamVerdict)) walk (branch AosWalkValue problem line code producers footer complete events unknown . (let unrestricted gaps = (aosSumGaps producers 0) in (let unrestricted unaccounted = (aosSumGaps producers 1) in (let unrestricted drops = (aosSumDrops producers) in (let unrestricted verdict = (lambda unrestricted status : Nat . (lambda unrestricted found : Nat . (constructor AosStreamVerdict AosStreamVerdictValue status found line code lines events unknown tailBytes gaps drops))) in (aosChoose (family AosStreamVerdict) problem (verdict 2 problem) (aosChoose (family AosStreamVerdict) (naturalIsZero lines) (verdict 2 1) (aosChoose (family AosStreamVerdict) tailBytes (verdict 1 9) (aosChoose (family AosStreamVerdict) (naturalIsZero footer) (verdict 1 7) (aosChoose (family AosStreamVerdict) (naturalIsZero complete) (verdict 1 8) (aosChoose (family AosStreamVerdict) unaccounted (verdict 1 10) (verdict 0 0)))))))))))))))) def aosStreamVerdict = (lambda unrestricted stream : Bytes . (let unrestricted segments = (aosSegments stream) in (aosFinishStreamScan (naturalSaturatingSubtract (aosListLength segments) 1) (bytes-length (aosListLast segments)) (aosWalkLines segments))))