Turn event streams into bounded evidence windows

A proactive agent reacts to observations that arrive without a new conversational prompt: a failed build, an updated document, a metric threshold, or a scheduled check. Separate the event from the action it might justify. An event saying a report changed can trigger a read and comparison; it does not automatically authorize publishing a new message. The event pipeline should first establish what happened, when it happened, and whether it is new information.

Use a consistent envelope with source, identifier, event type, subject, timestamp, and payload. CloudEvents standardizes common event metadata, including identity through the combination of source and ID. [CloudEvents specification](https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md). A shared envelope reduces adapter complexity, but it does not supply delivery guarantees, task policy, or trustworthy payload semantics. Validate those at the appropriate boundary before the agent reasons over the event.

Event time is when the underlying occurrence happened; processing time is when your system handled it. A monitor can receive an error recorded at 10:02 only at 10:07 after a network outage. Counting by arrival time makes it part of a different incident window than counting by occurrence time. Decide which interpretation matches the business question before building an aggregation.

Apache Beam's programming model discusses windows, triggers, watermarks, and late data as separate concepts. A watermark estimates event-time progress; it is not an absolute promise that older events can never arrive. [Apache Beam programming guide](https://beam.apache.org/documentation/programming-guide/). For a teaching simulation, an explicit cutoff is often enough. For a live system, document how late events update results, create corrections, or go to a separate review path so users can interpret changing summaries.

A fixed window of width sixty seconds can use the half-open interval [120, 180). An event at 120 belongs to it, while an event at 180 belongs to the next window. This convention prevents double counting at boundaries. Sliding windows overlap, so an event can legitimately contribute to several windows. Session windows group bursts separated by a gap and require different state than fixed buckets.

Suppose error events occur at 121, 140, 179, and 180. The [120, 180) window contains three errors. If a duplicate delivery of the 140 event arrives later, the count remains three after deduplication. If a genuinely new late event at 150 arrives after a report was issued, the count becomes four under an update policy. The decision to revise the report is distinct from the arithmetic used to compute the window.

Two events can have identical payload text and still represent separate occurrences. Conversely, one event can be delivered several times with transport metadata that differs across attempts. Deduplicate using the source's stable logical identifier and its scope. If event IDs are unique only within a tenant, include that tenant in the effective source identity. A hash of the entire payload is not a universal replacement because legitimate repeated events may share the same content.

Store a payload fingerprint with a seen ID when conflicting duplicates matter. Reusing the same logical ID with a different payload should trigger a diagnostic policy rather than silently changing history. The code uses first-seen identity for a bounded input list, which is a clear teaching contract. A persistent stream requires retention rules: forgetting IDs too quickly permits duplicates, while keeping every ID forever creates unbounded storage growth.

Deterministic stream operations should usually compute counts, sums, deduplication, and window membership before a model interprets the result. Asking a model to recount hundreds of raw events makes evaluation harder and can waste context on repeated boilerplate. Give it a compact evidence package containing the window definition, aggregate, representative examples, missing-data indicators, and relevant source references.

Keep enough lineage to reconstruct the aggregate. A report of twelve failures should identify the selected window and distinct event keys so a reviewer can audit the count. When the count changes due to late data, preserve the revision and explain the cause. The lesson and lab use finite synthetic lists, not a live CloudEvents or Beam runtime. Their purpose is to make time and identity contracts testable before those contracts are distributed across queues, workers, and model calls.

This teaching simulation uses a local event schema and a finite input list. It deduplicates by source and ID before applying a half-open event-time window. It is not a CloudEvents validator or a streaming engine and does not implement watermarks or persistent retention.