module Runtime.TelemetryPublication import Data.Bytes import Model.Parameter import Model.Word64 import Data.JSON import Observe.Interchange import Observe.Stream import Observe.Value import Runtime.ArenaCertificate import Runtime.NativeTelemetry import Runtime.TelemetryCapture import Runtime.TelemetryIdentity import Runtime.TelemetrySeal import Std.List import Std.Natural import Runtime.WordDigest -- ALPHATEL v2 capture batches reuse the maintained frame and SHA-256 owner. -- The identity field contains the four binary AOS identities. Header sequence -- is the batch ordinal; counters are the capture control snapshot. Payload is -- the prior published source sequence followed by occupied ordinary/reserved -- records, each carrying its original source sequence and eight exact u64s. -- Monotonic bytes are zero by contract and mean ABSENT, never measured zero. -- Status 0 keeps the producer open; 1 declares no further capture records. -- This says nothing about the workload's success or a device's clock. def publicationMaximumFrameBytes : Nat = 65536 def publicationEmptyCounters = (constructor NativeTelemetryCounters NativeTelemetryCountersValue 0 0 0 0 0 0 0 0) def publicationBodyPrefix : Bytes = (nativeTelemetryRecordBodyForVersion (byte 2) 0 0 (byte 0) (byte 0) (byte 0) (byte 0) publicationEmptyCounters (dataBytesRepeatByte (byte 0) telemetryIdentitySetBytes) b"" b"" 0 0) def publicationPayloadRecordsAt : Nat = (naturalAdd nativeTelemetryEmptyErrorPayloadAt captureWordBytes) def publicationBodyMaximumBytes = (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (naturalAdd publicationPayloadRecordsAt (captureQueueBytes ordinaryCapacity reservedCapacity)))) def publicationPaddedBytes = (lambda unrestricted bodyBytes : Nat . (naturalMultiply (naturalDivideUnchecked (naturalAdd bodyBytes (naturalAdd 1 (naturalAdd nativeTelemetryScalarBytes (naturalSaturatingSubtract 64 1)))) 64) 64)) def publicationRecordBytes = (lambda unrestricted bodyBytes : Nat . (naturalAdd (bytes-length nativeTelemetryMagic) (naturalAdd bodyBytes nativeTelemetryDigestBytes))) -- Reclaim requires both durability acknowledgements. A poisoned sink cannot -- be retried by appending behind its possibly partial tail. Recovery owns a -- new admitted sink/frontier; this primitive never silently discards a batch. def publicationReclaimAdmission = (lambda unrestricted version : Nat . (lambda unrestricted writeFull : Nat . (lambda unrestricted fileSynced : Nat . (lambda unrestricted directorySynced : Nat . (lambda unrestricted poisoned : Nat . (aosChoose Nat (naturalAnd (naturalEqual version 1) (naturalAnd (naturalLessOrEqual writeFull 1) (naturalAnd (naturalLessOrEqual fileSynced 1) (naturalAnd (naturalLessOrEqual directorySynced 1) (naturalLessOrEqual poisoned 1))))) (aosChoose Nat poisoned 6 (aosChoose Nat writeFull (aosChoose Nat (naturalAnd fileSynced directorySynced) 0 3) 2)) 4)))))) -- Capture counts are cumulative except occupied slots. Account for every -- selected source sequence since the prior durable frontier before encoding. def publicationConservation = (lambda unrestricted priorNext : Nat . (lambda unrestricted priorDropped : Nat . (lambda unrestricted priorExcluded : Nat . (lambda unrestricted next : Nat . (lambda unrestricted dropped : Nat . (lambda unrestricted excluded : Nat . (lambda unrestricted occupied : Nat . (naturalAnd (naturalLessOrEqual priorNext next) (naturalAnd (naturalLessOrEqual priorDropped dropped) (naturalAnd (naturalLessOrEqual priorExcluded excluded) (naturalAnd (naturalLessOrEqual (naturalSaturatingSubtract dropped priorDropped) (naturalSaturatingSubtract next priorNext)) (naturalEqual occupied (naturalSaturatingSubtract (naturalSaturatingSubtract next priorNext) (naturalSaturatingSubtract dropped priorDropped)))))))))))))) def publicationSequenceAdmission = (lambda unrestricted lower : Nat . (lambda unrestricted upper : Nat . (lambda unrestricted sequence : Nat . (naturalAnd (naturalLessOrEqual lower sequence) (naturalLess sequence upper))))) def publicationPublishedNextIndex : Nat = 0 def publicationPublishedNextAt : Nat = (naturalMultiply publicationPublishedNextIndex captureWordBytes) def publicationPublishedDroppedIndex : Nat = (succ publicationPublishedNextIndex) def publicationPublishedDroppedAt : Nat = (naturalMultiply publicationPublishedDroppedIndex captureWordBytes) def publicationPublishedExcludedIndex : Nat = (succ publicationPublishedDroppedIndex) def publicationPublishedExcludedAt : Nat = (naturalMultiply publicationPublishedExcludedIndex captureWordBytes) def publicationBatchOrdinalIndex : Nat = (succ publicationPublishedExcludedIndex) def publicationBatchOrdinalAt : Nat = (naturalMultiply publicationBatchOrdinalIndex captureWordBytes) def publicationFinishedIndex : Nat = (succ publicationBatchOrdinalIndex) def publicationFinishedAt : Nat = (naturalMultiply publicationFinishedIndex captureWordBytes) def publicationTerminalRequestIndex : Nat = (succ publicationFinishedIndex) def publicationTerminalRequestAt : Nat = (naturalMultiply publicationTerminalRequestIndex captureWordBytes) def publicationBodyLengthIndex : Nat = (succ publicationTerminalRequestIndex) def publicationBodyLengthAt : Nat = (naturalMultiply publicationBodyLengthIndex captureWordBytes) def publicationBlockCountIndex : Nat = (succ publicationBodyLengthIndex) def publicationBlockCountAt : Nat = (naturalMultiply publicationBlockCountIndex captureWordBytes) def publicationRecordLengthIndex : Nat = (succ publicationBlockCountIndex) def publicationRecordLengthAt : Nat = (naturalMultiply publicationRecordLengthIndex captureWordBytes) def publicationPoisonedIndex : Nat = (succ publicationRecordLengthIndex) def publicationPoisonedAt : Nat = (naturalMultiply publicationPoisonedIndex captureWordBytes) def publicationReadyIndex : Nat = (succ publicationPoisonedIndex) def publicationReadyAt : Nat = (naturalMultiply publicationReadyIndex captureWordBytes) def publicationSavedR12Index : Nat = (succ publicationReadyIndex) def publicationSavedR12At : Nat = (naturalMultiply publicationSavedR12Index captureWordBytes) def publicationSavedR13Index : Nat = (succ publicationSavedR12Index) def publicationSavedR13At : Nat = (naturalMultiply publicationSavedR13Index captureWordBytes) def publicationSavedR14Index : Nat = (succ publicationSavedR13Index) def publicationSavedR14At : Nat = (naturalMultiply publicationSavedR14Index captureWordBytes) def publicationSavedR15Index : Nat = (succ publicationSavedR14Index) def publicationSavedR15At : Nat = (naturalMultiply publicationSavedR15Index captureWordBytes) def publicationStateBytes : Nat = (naturalMultiply (succ publicationSavedR15Index) captureWordBytes) def publicationBodyAt : Nat = (captureAlign publicationStateBytes) -- The body bound is derived from the payload encoding. Scalar and typed drains -- share placement, frame limits, alignment and the same sealing scratch owner. def publicationRecordAtForBody = (lambda unrestricted bodyBytes : Nat . (naturalAdd publicationBodyAt (publicationPaddedBytes bodyBytes))) def publicationZeroClockAtForBody = (lambda unrestricted bodyBytes : Nat . (captureAlign (naturalAdd (publicationRecordAtForBody bodyBytes) (publicationRecordBytes bodyBytes)))) def publicationZeroClockBytes : Nat = (naturalMultiply 2 captureWordBytes) def publicationSealWorkAtForBody = (lambda unrestricted bodyBytes : Nat . (naturalAdd (publicationZeroClockAtForBody bodyBytes) publicationZeroClockBytes)) def publicationArenaBytesForBody = (lambda unrestricted bodyBytes : Nat . (naturalAdd (publicationSealWorkAtForBody bodyBytes) digestWorkBytes)) def publicationBase = captureWithIdentityArenaBytes def publicationWholeArenaBytesForBody = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (lambda unrestricted bodyBytes : Nat . (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) (publicationArenaBytesForBody bodyBytes)))))) def publicationWholeArenaResidentsForBody = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (lambda unrestricted bodyBytes : Nat . (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"publication-control" (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) 0) publicationStateBytes captureWordBytes) (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"publication-body" (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) publicationBodyAt) (publicationPaddedBytes bodyBytes) captureWordBytes) (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"publication-record" (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) (publicationRecordAtForBody bodyBytes)) (publicationRecordBytes bodyBytes) captureWordBytes) (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"publication-zero-clock" (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) (publicationZeroClockAtForBody bodyBytes)) publicationZeroClockBytes captureWordBytes) (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"publication-seal-work" (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) (publicationSealWorkAtForBody bodyBytes)) digestWorkBytes captureWordBytes) (captureWithIdentityArenaResidents applicationBytes ordinaryCapacity reservedCapacity)))))))))) def publicationStorageCertificateForBody = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (lambda unrestricted bodyBytes : Nat . (naturalAnd (naturalAnd (naturalEqual telemetryIdentitySetBytes nativeTelemetryDigestBytes) (naturalLessOrEqual publicationPayloadRecordsAt bodyBytes)) (naturalAnd (captureWithIdentityArenaCertificate applicationBytes ordinaryCapacity reservedCapacity) (naturalAnd (naturalLessOrEqual (publicationRecordBytes bodyBytes) publicationMaximumFrameBytes) (naturalAnd (naturalLessOrEqual (publicationWholeArenaBytesForBody applicationBytes ordinaryCapacity reservedCapacity bodyBytes) captureExtentMaximum) (arenaCertificate (publicationWholeArenaBytesForBody applicationBytes ordinaryCapacity reservedCapacity bodyBytes) (publicationWholeArenaResidentsForBody applicationBytes ordinaryCapacity reservedCapacity bodyBytes)))))))))) -- Scalar capture is one specialization of the shared bounded body layout. def publicationRecordAt = (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationRecordAtForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))) def publicationZeroClockAt = (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationZeroClockAtForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))) def publicationSealWorkAt = (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationSealWorkAtForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))) def publicationArenaBytes = (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationArenaBytesForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))) def publicationWholeArenaBytes = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationWholeArenaBytesForBody applicationBytes ordinaryCapacity reservedCapacity (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))) def publicationWholeArenaResidents = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationWholeArenaResidentsForBody applicationBytes ordinaryCapacity reservedCapacity (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))) def publicationStorageCertificate = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (publicationStorageCertificateForBody applicationBytes ordinaryCapacity reservedCapacity (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))) -- Version two carries the existing eight control words. A changed control -- shape requires its own admitted encoding, never silent field omission. def publicationArenaCertificate = (lambda unrestricted applicationBytes : Nat . (lambda unrestricted ordinaryCapacity : Nat . (lambda unrestricted reservedCapacity : Nat . (naturalAnd (naturalEqual captureControlBytes (bytes-length (nativeTelemetryEncodeCounters publicationEmptyCounters))) (publicationStorageCertificate applicationBytes ordinaryCapacity reservedCapacity))))) -- Version three is a bounded drain encoding for typed AOS records. Its -- identity, control snapshot, ordinal, frontier, digest and closure have the -- same owners as version two. The payload is the prior source frontier then -- LF-terminated descriptors/events. It does not reinterpret scalar words or -- format JSON in acquisition. Original origin/time/availability survive the -- drain; schema/stream admission remains Observe.Interchange/Observe.Stream. def publicationTypedRecordKey = (lambda unrestricted identities : Bytes . (bytes-append (dataBytesRenderHex (dataBytesTakeValidated telemetryIdentityBytes (dataBytesDropValidated (telemetryIdentityOffset telemetryAttemptIdentityIndex) identities))) (bytes-append b"/" (bytes-append (dataBytesRenderHex (dataBytesTakeValidated telemetryIdentityBytes (dataBytesDropValidated (telemetryIdentityOffset telemetryProducerIdentityIndex) identities))) (bytes-append b"/" (dataBytesRenderHex (dataBytesTakeValidated telemetryIdentityBytes (dataBytesDropValidated (telemetryIdentityOffset telemetryProducerEpochIndex) identities)))))))) def publicationTypedIdentitiesValid = (lambda unrestricted identities : Bytes . (naturalAnd (naturalEqual (bytes-length identities) telemetryIdentitySetBytes) (nat-eliminate (lambda unrestricted current : Nat . Nat) 1 (lambda unrestricted index : Nat . (lambda unrestricted induction : Nat . (naturalAnd induction (telemetryBinaryIdentityValid (dataBytesTakeValidated telemetryIdentityBytes (dataBytesDropValidated (telemetryIdentityOffset index) identities)))))) (succ telemetryProducerEpochIndex)))) def publicationTypedLineAdmission = (lambda unrestricted binding : Bytes . (lambda unrestricted lower : Nat . (lambda unrestricted upper : Nat . (lambda unrestricted line : Bytes . (naturalAnd (naturalIsZero (aosAnyByte (lambda unrestricted b : Byte . (byte-equal b (byte 10))) line)) (eliminate AosLine (lambda unrestricted current : (family AosLine) . Nat) (aosClassifyLine line) (branch AosLineInvalid code . 0) (branch AosLineValid record . (aosChoose Nat (bytes-equal (aosMemberBytes b"type" record) b"descriptor") (naturalIsZero (bytes-equal (aosMemberBytes b"kind" record) b"attempt")) (naturalAnd (bytes-equal (aosMemberBytes b"type" record) b"event") (let unrestricted identity = (aosMemberValue b"event_id" record) in (naturalAnd (bytes-equal binding (aosProducerKeyOf identity)) (publicationSequenceAdmission lower upper (aosDigitsValue (aosMemberBytes b"sequence" identity)))))))))))))) def publicationTypedRecordBytes = (lambda unrestricted records : (family StdList Bytes) . (bytes-builder-build (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . BytesBuilder) records (branch StdListEmpty . (bytes-builder-empty)) (branch StdListCons head tail induction . (bytes-builder-append (bytes-builder-chunk head) (bytes-builder-append (bytes-builder-chunk b"\n") induction)))))) def publicationTypedEventCount = (lambda unrestricted records : (family StdList Bytes) . (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . Nat) records (branch StdListEmpty . 0) (branch StdListCons head tail induction . (naturalAdd induction (eliminate AosLine (lambda unrestricted current : (family AosLine) . Nat) (aosClassifyLine head) (branch AosLineInvalid code . 0) (branch AosLineValid record . (bytes-equal (aosMemberBytes b"type" record) b"event"))))))) -- This checks individual record binding and the bounded payload population. -- Cross-record sequence uniqueness, cumulative conservation, descriptor -- immutability and stream schema meaning are additional admission gates; a -- successful result here is not a publication/durability acknowledgement. def publicationTypedPayloadAdmission = (lambda unrestricted identities : Bytes . (lambda unrestricted lower : Nat . (lambda unrestricted upper : Nat . (lambda unrestricted occupied : Nat . (lambda unrestricted records : (family StdList Bytes) . (naturalAnd (publicationTypedIdentitiesValid identities) (naturalAnd (naturalEqual (publicationTypedEventCount records) occupied) (naturalAnd (naturalLessOrEqual (aosListLength records) 1024) (naturalAnd (naturalLessOrEqual (publicationRecordBytes (naturalAdd publicationPayloadRecordsAt (bytes-length (publicationTypedRecordBytes records)))) publicationMaximumFrameBytes) (eliminate StdList (lambda unrestricted current : (family StdList Bytes) . Nat) records (branch StdListEmpty . 1) (branch StdListCons head tail induction . (naturalAnd induction (publicationTypedLineAdmission (publicationTypedRecordKey identities) lower upper head))))))))))))) -- A low-level body constructor, like nativeTelemetryRecordBodyForVersion. -- The caller must admit its control snapshot and typed payload before seal; -- the strict collector independently rechecks both against the original bytes. def publicationTypedBody = (lambda unrestricted ordinal : (family ModelWord64) . (lambda unrestricted terminal : Byte . (lambda unrestricted counters : (family NativeTelemetryCounters) . (lambda unrestricted identities : Bytes . (lambda unrestricted priorNext : (family ModelWord64) . (lambda unrestricted records : (family StdList Bytes) . (let unrestricted payload = (bytes-append (nativeTelemetryWord64Bytes priorNext) (publicationTypedRecordBytes records)) in (nativeTelemetryRecordBodyForVersion (byte 3) ordinal nativeTelemetryZeroWord64 (byte 0) (byte 0) (byte 0) terminal counters identities b"" payload nativeTelemetryZeroWord64 (modelWord64FromNaturalTruncated (bytes-length payload))))))))))