From e5aecee0deee993985bbd96081b83abb1f9949f7 Mon Sep 17 00:00:00 2001 From: Rijad Zuzo Date: Wed, 12 Aug 2026 22:59:59 +0200 Subject: [PATCH] feat: plan bounded event history rollups --- docs/REGIONAL_SIMULATION.md | 42 +- .../events/WorldEventDailyRollupStore.gd | 215 ++++++++ .../events/WorldEventDailyRollupStore.gd.uid | 1 + simulation/events/WorldEventRetentionPlan.gd | 486 ++++++++++++++++++ .../events/WorldEventRetentionPlan.gd.uid | 1 + .../events/WorldEventRetentionPlanner.gd | 435 ++++++++++++++++ .../events/WorldEventRetentionPlanner.gd.uid | 1 + .../state/WorldEventDailyRollupRecord.gd | 244 +++++++++ .../state/WorldEventDailyRollupRecord.gd.uid | 1 + tests/unit/test_world_event_retention.gd | 299 +++++++++++ tests/unit/test_world_event_retention.gd.uid | 1 + 11 files changed, 1722 insertions(+), 4 deletions(-) create mode 100644 simulation/events/WorldEventDailyRollupStore.gd create mode 100644 simulation/events/WorldEventDailyRollupStore.gd.uid create mode 100644 simulation/events/WorldEventRetentionPlan.gd create mode 100644 simulation/events/WorldEventRetentionPlan.gd.uid create mode 100644 simulation/events/WorldEventRetentionPlanner.gd create mode 100644 simulation/events/WorldEventRetentionPlanner.gd.uid create mode 100644 simulation/state/WorldEventDailyRollupRecord.gd create mode 100644 simulation/state/WorldEventDailyRollupRecord.gd.uid create mode 100644 tests/unit/test_world_event_retention.gd create mode 100644 tests/unit/test_world_event_retention.gd.uid diff --git a/docs/REGIONAL_SIMULATION.md b/docs/REGIONAL_SIMULATION.md index 5458eef..4b93e02 100644 --- a/docs/REGIONAL_SIMULATION.md +++ b/docs/REGIONAL_SIMULATION.md @@ -86,6 +86,39 @@ never-loaded, and load/unload executions must preserve cargo, headcount, facts, arrival tick, queue state, and checksum. Nearby visuals may interpolate route progress; they do not report elapsed travel completion. +## Event retention and daily rollups + +`WorldEventRetentionPlanner` is a pure, deterministic query over a complete +`WorldEventStore`. It returns a canonical `WorldEventRetentionPlan`; it never +deletes, replaces, or mutates source facts. The plan keeps the current day plus +the configured number of prior days raw. Only older event types explicitly +listed by the caller are eligible for aggregation. + +An eligible transaction must have a stable world and location plus either one +positive `item_id`/`amount` pair or a positive multi-item `cargo_ledger`. +`WorldEventDailyRollupRecord` groups those contributions by day, world, +location, event type, and item. It retains the event count, total amount, +tick/event-ID bounds, and a checksum of the exact contributing event IDs. +`WorldEventDailyRollupStore` provides canonical serialization and indexed +day/location/type/item queries. Incomplete or ambiguous transaction payloads +remain raw rather than being guessed into a rollup. + +Pinned fact IDs and caller-supplied causal fact IDs must exist in the source +store and always remain raw. Payload `cause_event_id` and `cause_event_ids` +references must resolve to prior facts; every referenced cause remains raw even +when its dependent transaction can be rolled up. Missing, dangling, or +non-prior causes fail the whole plan. Per-event reason traces, reciprocal causal +edges, source-ID fingerprints, and deterministic count telemetry survive the +plan's primitive-only round trip. + +This is a planning foundation, not an active retention authority. A future +integration must collect external references from knowledge, relationships, +situations, commitments, journals, and regional history; verify the plan's +source checksum immediately before applying it; and publish the retained raw +store, rollup store, and chunk manifest in one atomic transaction. A rollup is +aggregate history and must never satisfy a consumer that requires an exact +ordinary world fact. + ## Presentation relevance `PresentationRelevancePolicy` is a pure query over an active context and @@ -109,8 +142,9 @@ deterministic scheduler budgets, an empty final backlog, repeatable checksums, and measured construction/execution/serialization time. This is a headless data-and-scheduler baseline. It does not claim rendered -frame time, a complete market/economy, chunked persistence, or weak-PC GPU -performance. Those claims require separate active-world and hardware captures. +frame time, a complete market/economy, production-integrated chunked history +retention, or weak-PC GPU performance. Those claims require separate +active-world and hardware captures. ## Next integrations @@ -121,8 +155,8 @@ tests: save manifest; 2. migrate legacy global world-target registries to explicit active contexts; 3. make player and NPC work submit the same concrete action commands; -4. segment high-volume event history and save regional records in atomic, - lazily loaded chunks; +4. integrate the retention/rollup and chunk codecs as one atomic, + lazily loaded history transaction; 5. schedule 20 ordinary caravan trades across five settlement economies; 6. promote/demote presentation and population fidelity while pinning named and causally referenced people; diff --git a/simulation/events/WorldEventDailyRollupStore.gd b/simulation/events/WorldEventDailyRollupStore.gd new file mode 100644 index 0000000..7b2eab5 --- /dev/null +++ b/simulation/events/WorldEventDailyRollupStore.gd @@ -0,0 +1,215 @@ +class_name WorldEventDailyRollupStore +extends RefCounted + +const SCHEMA_VERSION := 1 +const RECORD_FIELDS := ["schema_version", "rollups"] + +var _records_by_key: Dictionary = {} +var _ordered_keys: Array[String] = [] +var _day_index: Dictionary = {} +var _location_index: Dictionary = {} +var _event_type_index: Dictionary = {} +var _item_index: Dictionary = {} + + +static func from_dictionary(store_data: Dictionary) -> WorldEventDailyRollupStore: + var store := WorldEventDailyRollupStore.new() + return store if store.restore_from_dictionary(store_data) else null + + +func size() -> int: + return _records_by_key.size() + + +func is_empty() -> bool: + return _records_by_key.is_empty() + + +func has_rollup( + day_index: int, + world_id: StringName, + location_id: StringName, + event_type: StringName, + item_id: StringName +) -> bool: + return _records_by_key.has( + WorldEventDailyRollupRecord.index_key_for( + day_index, world_id, location_id, event_type, item_id + ) + ) + + +func append(record: WorldEventDailyRollupRecord) -> bool: + return _append(record) + + +func get_by_key( + day_index: int, + world_id: StringName, + location_id: StringName, + event_type: StringName, + item_id: StringName +) -> WorldEventDailyRollupRecord: + var key := WorldEventDailyRollupRecord.index_key_for( + day_index, world_id, location_id, event_type, item_id + ) + return _records_by_key.get(key) as WorldEventDailyRollupRecord + + +func get_all() -> Array[WorldEventDailyRollupRecord]: + return _records_for_keys(_ordered_keys) + + +func get_for_day(day_index: int) -> Array[WorldEventDailyRollupRecord]: + return _records_for_keys(_day_index.get(day_index, [])) + + +func get_for_location( + world_id: StringName, location_id: StringName +) -> Array[WorldEventDailyRollupRecord]: + return _records_for_keys( + _location_index.get(WorldEventDailyRollupRecord.location_key_for(world_id, location_id), []) + ) + + +func get_for_event_type(event_type: StringName) -> Array[WorldEventDailyRollupRecord]: + return _records_for_keys(_event_type_index.get(event_type, [])) + + +func get_for_item(item_id: StringName) -> Array[WorldEventDailyRollupRecord]: + return _records_for_keys(_item_index.get(item_id, [])) + + +func restore(records: Array[WorldEventDailyRollupRecord]) -> bool: + var candidate := WorldEventDailyRollupStore.new() + for record: WorldEventDailyRollupRecord in records: + if not candidate._append(record): + return false + _adopt(candidate) + return true + + +func restore_from_dictionary(store_data: Dictionary) -> bool: + if not _has_exact_fields(store_data, RECORD_FIELDS): + return false + if ( + not store_data["schema_version"] is int + or int(store_data["schema_version"]) != SCHEMA_VERSION + or not store_data["rollups"] is Array + ): + return false + var records: Array[WorldEventDailyRollupRecord] = [] + for raw_record: Variant in store_data["rollups"]: + if not raw_record is Dictionary: + return false + var record := WorldEventDailyRollupRecord.from_dictionary(raw_record) + if record == null: + return false + records.append(record) + if not _records_are_canonical(records): + return false + return restore(records) + + +func to_dictionary() -> Dictionary: + var serialized: Array[Dictionary] = [] + for record: WorldEventDailyRollupRecord in get_all(): + serialized.append(record.to_dictionary()) + return { + "schema_version": SCHEMA_VERSION, + "rollups": serialized, + } + + +func checksum() -> String: + return JSON.stringify(to_dictionary()).sha256_text() + + +func _append(record: WorldEventDailyRollupRecord) -> bool: + if record == null or not record.is_valid() or _records_by_key.has(record.index_key()): + return false + var stored := WorldEventDailyRollupRecord.from_dictionary(record.to_dictionary()) + if stored == null: + return false + var key := stored.index_key() + _records_by_key[key] = stored + _insert_key(_ordered_keys, key) + _index_key(_day_index, stored.get_day_index(), key) + _index_key(_location_index, stored.location_index_key(), key) + _index_key(_event_type_index, stored.get_event_type(), key) + _index_key(_item_index, stored.get_item_id(), key) + return true + + +func _index_key(index: Dictionary, index_value: Variant, record_key: String) -> void: + if not index.has(index_value): + index[index_value] = [] + var keys: Array = index[index_value] + _insert_key(keys, record_key) + + +func _insert_key(keys: Array, record_key: String) -> void: + var low := 0 + var high := keys.size() + while low < high: + var middle := (low + high) / 2 + if _record_precedes(record_key, String(keys[middle])): + high = middle + else: + low = middle + 1 + keys.insert(low, record_key) + + +func _record_precedes(first_key: String, second_key: String) -> bool: + var first := _records_by_key[first_key] as WorldEventDailyRollupRecord + var second := _records_by_key[second_key] as WorldEventDailyRollupRecord + if first.get_day_index() != second.get_day_index(): + return first.get_day_index() < second.get_day_index() + if first.get_world_id() != second.get_world_id(): + return String(first.get_world_id()) < String(second.get_world_id()) + if first.get_location_id() != second.get_location_id(): + return String(first.get_location_id()) < String(second.get_location_id()) + if first.get_event_type() != second.get_event_type(): + return String(first.get_event_type()) < String(second.get_event_type()) + return String(first.get_item_id()) < String(second.get_item_id()) + + +func _records_for_keys(keys: Array) -> Array[WorldEventDailyRollupRecord]: + var records: Array[WorldEventDailyRollupRecord] = [] + for key: Variant in keys: + var record := _records_by_key.get(String(key)) as WorldEventDailyRollupRecord + if record != null: + records.append(record) + return records + + +func _records_are_canonical(records: Array[WorldEventDailyRollupRecord]) -> bool: + var candidate := WorldEventDailyRollupStore.new() + for record: WorldEventDailyRollupRecord in records: + if not candidate._append(record): + return false + var canonical_records := candidate.get_all() + if canonical_records.size() != records.size(): + return false + for index in range(records.size()): + if canonical_records[index].index_key() != records[index].index_key(): + return false + return true + + +func _adopt(source: WorldEventDailyRollupStore) -> void: + _records_by_key = source._records_by_key + _ordered_keys = source._ordered_keys + _day_index = source._day_index + _location_index = source._location_index + _event_type_index = source._event_type_index + _item_index = source._item_index + + +static func _has_exact_fields(record_data: Dictionary, fields: Array) -> bool: + if record_data.size() != fields.size(): + return false + for field: String in fields: + if not record_data.has(field): + return false + return true diff --git a/simulation/events/WorldEventDailyRollupStore.gd.uid b/simulation/events/WorldEventDailyRollupStore.gd.uid new file mode 100644 index 0000000..0e51253 --- /dev/null +++ b/simulation/events/WorldEventDailyRollupStore.gd.uid @@ -0,0 +1 @@ +uid://p61id8fes4wk diff --git a/simulation/events/WorldEventRetentionPlan.gd b/simulation/events/WorldEventRetentionPlan.gd new file mode 100644 index 0000000..ef74439 --- /dev/null +++ b/simulation/events/WorldEventRetentionPlan.gd @@ -0,0 +1,486 @@ +class_name WorldEventRetentionPlan +extends RefCounted + +const SCHEMA_VERSION := 1 +const DECISION_RETAIN_RAW := "retain_raw" +const DECISION_ROLLED_UP := "rolled_up" +const RECORD_FIELDS := [ + "schema_version", + "current_tick", + "ticks_per_day", + "raw_retention_days", + "first_raw_day_index", + "source_checksum", + "high_volume_event_types", + "pinned_fact_ids", + "causal_fact_ids", + "retained_event_ids", + "rolled_up_event_ids", + "rollup_store", + "reason_traces", + "telemetry", +] +const TRACE_FIELDS := [ + "event_id", + "decision", + "reasons", + "cause_event_ids", + "required_by_event_ids", + "rollup_keys", +] +const TELEMETRY_FIELDS := [ + "succeeded", + "input_event_count", + "retained_event_count", + "rolled_up_event_count", + "rollup_record_count", + "rollup_contribution_count", + "pinned_fact_count", + "causal_fact_count", + "causal_ancestor_count", + "recent_event_count", + "non_rollup_event_count", + "invalid_rollup_event_count", + "first_raw_day_index", + "source_checksum", + "source_unchanged", +] + +var _data: Dictionary = {} +var _rollup_store := WorldEventDailyRollupStore.new() +var _traces_by_event_id: Dictionary = {} + + +static func create( + current_tick: int, + ticks_per_day: int, + raw_retention_days: int, + first_raw_day_index: int, + source_checksum: String, + high_volume_event_types: Array[StringName], + pinned_fact_ids: Array[int], + causal_fact_ids: Array[int], + retained_event_ids: Array[int], + rolled_up_event_ids: Array[int], + rollup_store: WorldEventDailyRollupStore, + reason_traces: Array[Dictionary], + telemetry: Dictionary +) -> WorldEventRetentionPlan: + if rollup_store == null: + return null + var serialized_event_types: Array[String] = [] + for event_type: StringName in high_volume_event_types: + serialized_event_types.append(String(event_type)) + return from_dictionary( + { + "schema_version": SCHEMA_VERSION, + "current_tick": current_tick, + "ticks_per_day": ticks_per_day, + "raw_retention_days": raw_retention_days, + "first_raw_day_index": first_raw_day_index, + "source_checksum": source_checksum, + "high_volume_event_types": serialized_event_types, + "pinned_fact_ids": pinned_fact_ids.duplicate(), + "causal_fact_ids": causal_fact_ids.duplicate(), + "retained_event_ids": retained_event_ids.duplicate(), + "rolled_up_event_ids": rolled_up_event_ids.duplicate(), + "rollup_store": rollup_store.to_dictionary(), + "reason_traces": reason_traces.duplicate(true), + "telemetry": telemetry.duplicate(true), + } + ) + + +static func from_dictionary(record_data: Dictionary) -> WorldEventRetentionPlan: + if not _has_exact_fields(record_data, RECORD_FIELDS): + return null + if ( + not record_data["schema_version"] is int + or int(record_data["schema_version"]) != SCHEMA_VERSION + or not record_data["current_tick"] is int + or not record_data["ticks_per_day"] is int + or not record_data["raw_retention_days"] is int + or not record_data["first_raw_day_index"] is int + or not record_data["source_checksum"] is String + or not record_data["high_volume_event_types"] is Array + or not record_data["pinned_fact_ids"] is Array + or not record_data["causal_fact_ids"] is Array + or not record_data["retained_event_ids"] is Array + or not record_data["rolled_up_event_ids"] is Array + or not record_data["rollup_store"] is Dictionary + or not record_data["reason_traces"] is Array + or not record_data["telemetry"] is Dictionary + ): + return null + var event_types: Variant = _normalize_string_array(record_data["high_volume_event_types"]) + var pinned_ids: Variant = _normalize_int_array(record_data["pinned_fact_ids"]) + var causal_ids: Variant = _normalize_int_array(record_data["causal_fact_ids"]) + var retained_ids: Variant = _normalize_int_array(record_data["retained_event_ids"]) + var rolled_up_ids: Variant = _normalize_int_array(record_data["rolled_up_event_ids"]) + if ( + event_types == null + or pinned_ids == null + or causal_ids == null + or retained_ids == null + or rolled_up_ids == null + or event_types != record_data["high_volume_event_types"] + or pinned_ids != record_data["pinned_fact_ids"] + or causal_ids != record_data["causal_fact_ids"] + or retained_ids != record_data["retained_event_ids"] + or rolled_up_ids != record_data["rolled_up_event_ids"] + ): + return null + var current_tick := int(record_data["current_tick"]) + var ticks_per_day := int(record_data["ticks_per_day"]) + var raw_retention_days := int(record_data["raw_retention_days"]) + var first_raw_day_index := int(record_data["first_raw_day_index"]) + if ( + current_tick < 0 + or ticks_per_day <= 0 + or raw_retention_days < 0 + or (first_raw_day_index != maxi(int(current_tick / ticks_per_day) - raw_retention_days, 0)) + or not _is_sha256(String(record_data["source_checksum"])) + ): + return null + var rollup_store := WorldEventDailyRollupStore.from_dictionary(record_data["rollup_store"]) + if rollup_store == null: + return null + var normalized_traces: Variant = _normalize_traces(record_data["reason_traces"]) + if normalized_traces == null or normalized_traces != record_data["reason_traces"]: + return null + var telemetry: Dictionary = record_data["telemetry"] + if not _telemetry_is_valid(telemetry, String(record_data["source_checksum"])): + return null + if not _cross_references_are_valid( + pinned_ids, + causal_ids, + retained_ids, + rolled_up_ids, + rollup_store, + normalized_traces, + telemetry, + first_raw_day_index + ): + return null + var plan := WorldEventRetentionPlan.new() + plan._data = { + "current_tick": current_tick, + "ticks_per_day": ticks_per_day, + "raw_retention_days": raw_retention_days, + "first_raw_day_index": first_raw_day_index, + "source_checksum": String(record_data["source_checksum"]), + "high_volume_event_types": event_types, + "pinned_fact_ids": pinned_ids, + "causal_fact_ids": causal_ids, + "retained_event_ids": retained_ids, + "rolled_up_event_ids": rolled_up_ids, + "reason_traces": normalized_traces, + "telemetry": telemetry.duplicate(true), + } + plan._rollup_store = rollup_store + for trace: Dictionary in normalized_traces: + plan._traces_by_event_id[int(trace["event_id"])] = trace.duplicate(true) + return plan + + +func get_retained_event_ids() -> Array[int]: + return _copy_int_array(_data["retained_event_ids"]) + + +func get_rolled_up_event_ids() -> Array[int]: + return _copy_int_array(_data["rolled_up_event_ids"]) + + +func get_rollup_store() -> WorldEventDailyRollupStore: + return WorldEventDailyRollupStore.from_dictionary(_rollup_store.to_dictionary()) + + +func get_reason_trace(event_id: int) -> Dictionary: + var trace: Dictionary = _traces_by_event_id.get(event_id, {} as Dictionary) + return trace.duplicate(true) + + +func get_reason_traces() -> Array[Dictionary]: + var traces: Array[Dictionary] = [] + for trace: Dictionary in _data["reason_traces"]: + traces.append(trace.duplicate(true)) + return traces + + +func get_telemetry() -> Dictionary: + return (_data["telemetry"] as Dictionary).duplicate(true) + + +func get_source_checksum() -> String: + return String(_data["source_checksum"]) + + +func to_dictionary() -> Dictionary: + return { + "schema_version": SCHEMA_VERSION, + "current_tick": int(_data["current_tick"]), + "ticks_per_day": int(_data["ticks_per_day"]), + "raw_retention_days": int(_data["raw_retention_days"]), + "first_raw_day_index": int(_data["first_raw_day_index"]), + "source_checksum": String(_data["source_checksum"]), + "high_volume_event_types": (_data["high_volume_event_types"] as Array).duplicate(), + "pinned_fact_ids": (_data["pinned_fact_ids"] as Array).duplicate(), + "causal_fact_ids": (_data["causal_fact_ids"] as Array).duplicate(), + "retained_event_ids": (_data["retained_event_ids"] as Array).duplicate(), + "rolled_up_event_ids": (_data["rolled_up_event_ids"] as Array).duplicate(), + "rollup_store": _rollup_store.to_dictionary(), + "reason_traces": (_data["reason_traces"] as Array).duplicate(true), + "telemetry": (_data["telemetry"] as Dictionary).duplicate(true), + } + + +func checksum() -> String: + return JSON.stringify(to_dictionary()).sha256_text() + + +static func _normalize_traces(raw_traces: Array) -> Variant: + var traces: Array[Dictionary] = [] + var seen: Dictionary = {} + var last_event_id := -1 + for raw_trace: Variant in raw_traces: + if not raw_trace is Dictionary or not _has_exact_fields(raw_trace, TRACE_FIELDS): + return null + if ( + not raw_trace["event_id"] is int + or not raw_trace["decision"] is String + or not raw_trace["reasons"] is Array + or not raw_trace["cause_event_ids"] is Array + or not raw_trace["required_by_event_ids"] is Array + or not raw_trace["rollup_keys"] is Array + ): + return null + var event_id := int(raw_trace["event_id"]) + var decision := String(raw_trace["decision"]) + var reasons: Variant = _normalize_string_array(raw_trace["reasons"]) + var cause_ids: Variant = _normalize_int_array(raw_trace["cause_event_ids"]) + var required_by_ids: Variant = _normalize_int_array(raw_trace["required_by_event_ids"]) + var rollup_keys: Variant = _normalize_string_array(raw_trace["rollup_keys"]) + if ( + event_id < 0 + or event_id <= last_event_id + or seen.has(event_id) + or decision not in [DECISION_RETAIN_RAW, DECISION_ROLLED_UP] + or reasons == null + or reasons.is_empty() + or cause_ids == null + or required_by_ids == null + or rollup_keys == null + or reasons != raw_trace["reasons"] + or cause_ids != raw_trace["cause_event_ids"] + or required_by_ids != raw_trace["required_by_event_ids"] + or rollup_keys != raw_trace["rollup_keys"] + ): + return null + seen[event_id] = true + last_event_id = event_id + traces.append(raw_trace.duplicate(true)) + return traces + + +static func _telemetry_is_valid(telemetry: Dictionary, source_checksum: String) -> bool: + if not _has_exact_fields(telemetry, TELEMETRY_FIELDS): + return false + if ( + telemetry["succeeded"] is not bool + or not bool(telemetry["succeeded"]) + or telemetry["source_unchanged"] is not bool + or not bool(telemetry["source_unchanged"]) + or telemetry["source_checksum"] is not String + or String(telemetry["source_checksum"]) != source_checksum + ): + return false + for field: String in TELEMETRY_FIELDS: + if field in ["succeeded", "source_unchanged", "source_checksum"]: + continue + if telemetry[field] is not int or int(telemetry[field]) < 0: + return false + return true + + +static func _cross_references_are_valid( + pinned_ids: Array, + causal_ids: Array, + retained_ids: Array, + rolled_up_ids: Array, + rollup_store: WorldEventDailyRollupStore, + traces: Array, + telemetry: Dictionary, + first_raw_day_index: int +) -> bool: + var retained_set := _id_set(retained_ids) + var rolled_up_set := _id_set(rolled_up_ids) + var traces_by_id: Dictionary = {} + for trace: Dictionary in traces: + traces_by_id[int(trace["event_id"])] = trace + for event_id: int in retained_ids: + if rolled_up_set.has(event_id): + return false + for event_id: int in pinned_ids: + if ( + not retained_set.has(event_id) + or not traces_by_id.has(event_id) + or "pinned_fact" not in (traces_by_id[event_id] as Dictionary)["reasons"] + ): + return false + for event_id: int in causal_ids: + if ( + not retained_set.has(event_id) + or not traces_by_id.has(event_id) + or "explicit_causal_fact" not in (traces_by_id[event_id] as Dictionary)["reasons"] + ): + return false + var rollup_keys: Dictionary = {} + var contribution_count := 0 + for record: WorldEventDailyRollupRecord in rollup_store.get_all(): + rollup_keys[record.index_key()] = true + contribution_count += record.get_event_count() + var trace_ids: Dictionary = {} + var causal_ancestor_count := 0 + var recent_count := 0 + var non_rollup_count := 0 + var invalid_rollup_count := 0 + var source_ids_by_rollup_key: Dictionary = {} + for trace: Dictionary in traces: + var event_id := int(trace["event_id"]) + trace_ids[event_id] = true + var is_retained := retained_set.has(event_id) + var is_rolled_up := rolled_up_set.has(event_id) + if is_retained == is_rolled_up: + return false + if is_retained and String(trace["decision"]) != DECISION_RETAIN_RAW: + return false + if is_rolled_up and String(trace["decision"]) != DECISION_ROLLED_UP: + return false + var reasons: Array = trace["reasons"] + var trace_rollup_keys: Array = trace["rollup_keys"] + if is_retained and not trace_rollup_keys.is_empty(): + return false + if is_retained and "rolled_up_daily" in reasons: + return false + if is_rolled_up and (reasons != ["rolled_up_daily"] or trace_rollup_keys.is_empty()): + return false + for rollup_key: String in trace_rollup_keys: + if not rollup_keys.has(rollup_key): + return false + if not source_ids_by_rollup_key.has(rollup_key): + source_ids_by_rollup_key[rollup_key] = [] + var source_ids: Array = source_ids_by_rollup_key[rollup_key] + source_ids.append(event_id) + for cause_id: int in trace["cause_event_ids"]: + if not retained_set.has(cause_id) or not traces_by_id.has(cause_id): + return false + var cause_trace: Dictionary = traces_by_id[cause_id] + if ( + "causal_ancestor" not in cause_trace["reasons"] + or event_id not in cause_trace["required_by_event_ids"] + ): + return false + for required_by_id: int in trace["required_by_event_ids"]: + if not traces_by_id.has(required_by_id): + return false + var required_by_trace: Dictionary = traces_by_id[required_by_id] + if event_id not in required_by_trace["cause_event_ids"]: + return false + if "causal_ancestor" in reasons: + causal_ancestor_count += 1 + if "within_raw_retention" in reasons: + recent_count += 1 + if "not_rollup_eligible" in reasons: + non_rollup_count += 1 + if "rollup_payload_invalid" in reasons: + invalid_rollup_count += 1 + if trace_ids.size() != retained_ids.size() + rolled_up_ids.size(): + return false + if source_ids_by_rollup_key.size() != rollup_store.size(): + return false + for record: WorldEventDailyRollupRecord in rollup_store.get_all(): + var rollup_key := record.index_key() + if not source_ids_by_rollup_key.has(rollup_key): + return false + var source_ids: Array = source_ids_by_rollup_key[rollup_key] + source_ids.sort() + if ( + record.get_event_count() != source_ids.size() + or record.get_minimum_event_id() != int(source_ids.front()) + or record.get_maximum_event_id() != int(source_ids.back()) + or record.get_source_event_ids_checksum() != JSON.stringify(source_ids).sha256_text() + ): + return false + return ( + int(telemetry["input_event_count"]) == trace_ids.size() + and int(telemetry["retained_event_count"]) == retained_ids.size() + and int(telemetry["rolled_up_event_count"]) == rolled_up_ids.size() + and int(telemetry["rollup_record_count"]) == rollup_store.size() + and int(telemetry["rollup_contribution_count"]) == contribution_count + and int(telemetry["pinned_fact_count"]) == pinned_ids.size() + and int(telemetry["causal_fact_count"]) == causal_ids.size() + and int(telemetry["causal_ancestor_count"]) == causal_ancestor_count + and int(telemetry["recent_event_count"]) == recent_count + and int(telemetry["non_rollup_event_count"]) == non_rollup_count + and int(telemetry["invalid_rollup_event_count"]) == invalid_rollup_count + and int(telemetry["first_raw_day_index"]) == first_raw_day_index + ) + + +static func _normalize_string_array(values: Array) -> Variant: + var seen: Dictionary = {} + var normalized: Array[String] = [] + for value: Variant in values: + if (not value is String and not value is StringName) or String(value).is_empty(): + return null + var text := String(value) + if seen.has(text): + return null + seen[text] = true + normalized.append(text) + normalized.sort() + return normalized + + +static func _normalize_int_array(values: Array) -> Variant: + var seen: Dictionary = {} + var normalized: Array[int] = [] + for value: Variant in values: + if not value is int or int(value) < 0 or seen.has(int(value)): + return null + seen[int(value)] = true + normalized.append(int(value)) + normalized.sort() + return normalized + + +static func _id_set(values: Array) -> Dictionary: + var result: Dictionary = {} + for value: int in values: + result[value] = true + return result + + +static func _copy_int_array(values: Array) -> Array[int]: + var copied: Array[int] = [] + for value: int in values: + copied.append(value) + return copied + + +static func _has_exact_fields(record_data: Dictionary, fields: Array) -> bool: + if record_data.size() != fields.size(): + return false + for field: String in fields: + if not record_data.has(field): + return false + return true + + +static func _is_sha256(value: String) -> bool: + if value.length() != 64: + return false + for character_index in range(value.length()): + if value[character_index] not in "0123456789abcdef": + return false + return true diff --git a/simulation/events/WorldEventRetentionPlan.gd.uid b/simulation/events/WorldEventRetentionPlan.gd.uid new file mode 100644 index 0000000..f151ee1 --- /dev/null +++ b/simulation/events/WorldEventRetentionPlan.gd.uid @@ -0,0 +1 @@ +uid://bj0uo56quas27 diff --git a/simulation/events/WorldEventRetentionPlanner.gd b/simulation/events/WorldEventRetentionPlanner.gd new file mode 100644 index 0000000..52cce53 --- /dev/null +++ b/simulation/events/WorldEventRetentionPlanner.gd @@ -0,0 +1,435 @@ +class_name WorldEventRetentionPlanner +extends RefCounted + +const REASON_PINNED_FACT := "pinned_fact" +const REASON_EXPLICIT_CAUSAL_FACT := "explicit_causal_fact" +const REASON_CAUSAL_ANCESTOR := "causal_ancestor" +const REASON_WITHIN_RAW_RETENTION := "within_raw_retention" +const REASON_NOT_ROLLUP_ELIGIBLE := "not_rollup_eligible" +const REASON_ROLLUP_PAYLOAD_INVALID := "rollup_payload_invalid" +const REASON_ROLLED_UP_DAILY := "rolled_up_daily" + +var _last_error := "" +var _last_telemetry: Dictionary = {} + + +func plan( + source: WorldEventStore, + current_tick: int, + ticks_per_day: int, + raw_retention_days: int, + high_volume_event_types: Array, + pinned_fact_ids: Array = [], + causal_fact_ids: Array = [] +) -> WorldEventRetentionPlan: + _last_error = "" + _last_telemetry = { + "succeeded": false, + "input_event_count": source.size() if source != null else 0, + "error": "", + } + if source == null: + return _fail("source WorldEventStore is null") + if current_tick < 0 or ticks_per_day <= 0 or raw_retention_days < 0: + return _fail("retention time configuration is invalid") + var normalized_types: Variant = _normalize_event_types(high_volume_event_types) + var normalized_pins: Variant = _normalize_event_ids(pinned_fact_ids) + var normalized_causal_facts: Variant = _normalize_event_ids(causal_fact_ids) + if normalized_types == null: + return _fail("high-volume event types are invalid or duplicated") + if normalized_pins == null: + return _fail("pinned fact IDs are invalid or duplicated") + if normalized_causal_facts == null: + return _fail("causal fact IDs are invalid or duplicated") + for event_id: int in normalized_pins: + if not source.has_event(event_id): + return _fail("pinned fact ID %d is missing from the source store" % event_id) + for event_id: int in normalized_causal_facts: + if not source.has_event(event_id): + return _fail("causal fact ID %d is missing from the source store" % event_id) + var source_serialized := JSON.stringify(source.to_dictionary()) + var source_checksum := source_serialized.sha256_text() + var events := source.get_all() + var events_by_id: Dictionary = {} + var order_by_id: Dictionary = {} + var traces_by_id: Dictionary = {} + for event_index in range(events.size()): + var event := events[event_index] as WorldEventRecord + if event == null or event.get_tick() > current_tick: + return _fail("source contains a null or future event") + var event_id := event.get_event_id() + events_by_id[event_id] = event + order_by_id[event_id] = event_index + var cause_ids: Variant = _extract_cause_event_ids(event.get_payload()) + if cause_ids == null: + return _fail("event %d has an invalid causal fact contract" % event_id) + traces_by_id[event_id] = { + "event_id": event_id, + "decision": "", + "reasons": [], + "cause_event_ids": cause_ids, + "required_by_event_ids": [], + "rollup_keys": [], + } + for event: WorldEventRecord in events: + var event_id := event.get_event_id() + var trace: Dictionary = traces_by_id[event_id] + for cause_event_id: int in trace["cause_event_ids"]: + if not events_by_id.has(cause_event_id): + return _fail( + "event %d references missing causal fact %d" % [event_id, cause_event_id] + ) + if int(order_by_id[cause_event_id]) >= int(order_by_id[event_id]): + return _fail( + "event %d references non-prior causal fact %d" % [event_id, cause_event_id] + ) + var high_volume_type_set: Dictionary = {} + for event_type: StringName in normalized_types: + high_volume_type_set[String(event_type)] = true + var current_day_index := int(current_tick / ticks_per_day) + var first_raw_day_index := maxi(current_day_index - raw_retention_days, 0) + var retained: Dictionary = {} + var contributions_by_event_id: Dictionary = {} + for event: WorldEventRecord in events: + var event_id := event.get_event_id() + var event_day_index := int(event.get_tick() / ticks_per_day) + if event_day_index >= first_raw_day_index: + _mark_retained(retained, traces_by_id, event_id, REASON_WITHIN_RAW_RETENTION) + continue + if not high_volume_type_set.has(String(event.get_event_type())): + _mark_retained(retained, traces_by_id, event_id, REASON_NOT_ROLLUP_ELIGIBLE) + continue + var contributions: Variant = _extract_rollup_contributions(event, event_day_index) + if contributions == null: + _mark_retained(retained, traces_by_id, event_id, REASON_ROLLUP_PAYLOAD_INVALID) + continue + contributions_by_event_id[event_id] = contributions + for event_id: int in normalized_pins: + _mark_retained(retained, traces_by_id, event_id, REASON_PINNED_FACT) + for event_id: int in normalized_causal_facts: + _mark_retained(retained, traces_by_id, event_id, REASON_EXPLICIT_CAUSAL_FACT) + for event: WorldEventRecord in events: + var required_by_id := event.get_event_id() + var child_trace: Dictionary = traces_by_id[required_by_id] + for cause_event_id: int in child_trace["cause_event_ids"]: + _mark_retained(retained, traces_by_id, cause_event_id, REASON_CAUSAL_ANCESTOR) + var cause_trace: Dictionary = traces_by_id[cause_event_id] + var required_by_ids: Array = cause_trace["required_by_event_ids"] + if required_by_id not in required_by_ids: + required_by_ids.append(required_by_id) + var accumulators: Dictionary = {} + var rolled_up: Dictionary = {} + for event: WorldEventRecord in events: + var event_id := event.get_event_id() + if retained.has(event_id): + continue + if not contributions_by_event_id.has(event_id): + return _fail("an event selected for rollup has no validated contribution") + var trace: Dictionary = traces_by_id[event_id] + _add_reason(trace, REASON_ROLLED_UP_DAILY) + for contribution: Dictionary in contributions_by_event_id[event_id]: + var rollup_key := WorldEventDailyRollupRecord.index_key_for( + int(contribution["day_index"]), + StringName(contribution["world_id"]), + StringName(contribution["location_id"]), + StringName(contribution["event_type"]), + StringName(contribution["item_id"]) + ) + if not accumulators.has(rollup_key): + accumulators[rollup_key] = { + "day_index": int(contribution["day_index"]), + "world_id": String(contribution["world_id"]), + "location_id": String(contribution["location_id"]), + "event_type": String(contribution["event_type"]), + "item_id": String(contribution["item_id"]), + "source_event_ids": [], + "total_amount": 0.0, + "minimum_tick": event.get_tick(), + "maximum_tick": event.get_tick(), + } + var accumulator: Dictionary = accumulators[rollup_key] + var source_event_ids: Array = accumulator["source_event_ids"] + source_event_ids.append(event_id) + accumulator["total_amount"] = ( + float(accumulator["total_amount"]) + float(contribution["amount"]) + ) + accumulator["minimum_tick"] = mini(int(accumulator["minimum_tick"]), event.get_tick()) + accumulator["maximum_tick"] = maxi(int(accumulator["maximum_tick"]), event.get_tick()) + var trace_rollup_keys: Array = trace["rollup_keys"] + trace_rollup_keys.append(rollup_key) + rolled_up[event_id] = true + var rollup_store := WorldEventDailyRollupStore.new() + var accumulator_keys: Array = accumulators.keys() + accumulator_keys.sort() + for rollup_key: String in accumulator_keys: + var accumulator: Dictionary = accumulators[rollup_key] + var record := WorldEventDailyRollupRecord.create( + int(accumulator["day_index"]), + StringName(accumulator["world_id"]), + StringName(accumulator["location_id"]), + StringName(accumulator["event_type"]), + StringName(accumulator["item_id"]), + _copy_int_array(accumulator["source_event_ids"]), + float(accumulator["total_amount"]), + int(accumulator["minimum_tick"]), + int(accumulator["maximum_tick"]) + ) + if record == null or not rollup_store.append(record): + return _fail("daily rollup record construction failed") + var retained_ids := _sorted_int_keys(retained) + var rolled_up_ids := _sorted_int_keys(rolled_up) + var trace_event_ids := _sorted_int_keys(traces_by_id) + var reason_traces: Array[Dictionary] = [] + for event_id: int in trace_event_ids: + var trace: Dictionary = traces_by_id[event_id] + var reasons: Array = trace["reasons"] + var required_by_ids: Array = trace["required_by_event_ids"] + var rollup_keys: Array = trace["rollup_keys"] + reasons.sort() + required_by_ids.sort() + rollup_keys.sort() + trace["decision"] = ( + WorldEventRetentionPlan.DECISION_RETAIN_RAW + if retained.has(event_id) + else WorldEventRetentionPlan.DECISION_ROLLED_UP + ) + reason_traces.append(trace.duplicate(true)) + var source_unchanged := JSON.stringify(source.to_dictionary()) == source_serialized + if not source_unchanged: + return _fail("source WorldEventStore changed while planning") + var telemetry := { + "succeeded": true, + "input_event_count": events.size(), + "retained_event_count": retained_ids.size(), + "rolled_up_event_count": rolled_up_ids.size(), + "rollup_record_count": rollup_store.size(), + "rollup_contribution_count": _rollup_contribution_count(rollup_store), + "pinned_fact_count": normalized_pins.size(), + "causal_fact_count": normalized_causal_facts.size(), + "causal_ancestor_count": _reason_count(reason_traces, REASON_CAUSAL_ANCESTOR), + "recent_event_count": _reason_count(reason_traces, REASON_WITHIN_RAW_RETENTION), + "non_rollup_event_count": _reason_count(reason_traces, REASON_NOT_ROLLUP_ELIGIBLE), + "invalid_rollup_event_count": _reason_count(reason_traces, REASON_ROLLUP_PAYLOAD_INVALID), + "first_raw_day_index": first_raw_day_index, + "source_checksum": source_checksum, + "source_unchanged": source_unchanged, + } + var result := WorldEventRetentionPlan.create( + current_tick, + ticks_per_day, + raw_retention_days, + first_raw_day_index, + source_checksum, + _copy_string_name_array(normalized_types), + _copy_int_array(normalized_pins), + _copy_int_array(normalized_causal_facts), + retained_ids, + rolled_up_ids, + rollup_store, + reason_traces, + telemetry + ) + if result == null: + return _fail("retention plan validation failed") + _last_telemetry = telemetry.duplicate(true) + return result + + +func get_last_error() -> String: + return _last_error + + +func get_last_telemetry() -> Dictionary: + return _last_telemetry.duplicate(true) + + +func _fail(message: String) -> WorldEventRetentionPlan: + _last_error = message + _last_telemetry["error"] = message + _last_telemetry["succeeded"] = false + return null + + +static func _mark_retained( + retained: Dictionary, traces_by_id: Dictionary, event_id: int, reason: String +) -> void: + retained[event_id] = true + var trace: Dictionary = traces_by_id[event_id] + _add_reason(trace, reason) + + +static func _add_reason(trace: Dictionary, reason: String) -> void: + var reasons: Array = trace["reasons"] + if reason not in reasons: + reasons.append(reason) + + +static func _extract_cause_event_ids(payload: Dictionary) -> Variant: + var seen: Dictionary = {} + var event_ids: Array[int] = [] + if payload.has("cause_event_id"): + var single_cause: Variant = payload["cause_event_id"] + if not single_cause is int or int(single_cause) < -1: + return null + if int(single_cause) >= 0: + seen[int(single_cause)] = true + event_ids.append(int(single_cause)) + if payload.has("cause_event_ids"): + var multiple_causes: Variant = payload["cause_event_ids"] + if not multiple_causes is Array: + return null + for raw_cause: Variant in multiple_causes: + if not raw_cause is int or int(raw_cause) < 0: + return null + if seen.has(int(raw_cause)): + continue + seen[int(raw_cause)] = true + event_ids.append(int(raw_cause)) + event_ids.sort() + return event_ids + + +static func _extract_rollup_contributions(event: WorldEventRecord, day_index: int) -> Variant: + if event == null or not event.has_location(): + return null + var location := event.get_location() + if ( + location == null + or location.get_world_id().is_empty() + or location.get_location_id().is_empty() + ): + return null + var payload := event.get_payload() + var has_item_amount := payload.has("item_id") or payload.has("amount") + var has_cargo_ledger := payload.has("cargo_ledger") + if has_item_amount and has_cargo_ledger: + return null + var contributions: Array[Dictionary] = [] + if has_cargo_ledger: + if not payload["cargo_ledger"] is Dictionary: + return null + var ledger: Dictionary = payload["cargo_ledger"] + var source_keys: Dictionary = {} + for raw_item_id: Variant in ledger: + if not raw_item_id is String and not raw_item_id is StringName: + return null + var item_id := String(raw_item_id) + if item_id.is_empty() or source_keys.has(item_id): + return null + var raw_amount: Variant = ledger[raw_item_id] + if ( + not raw_amount is int and not raw_amount is float + or not is_finite(float(raw_amount)) + or float(raw_amount) <= 0.0 + ): + return null + source_keys[item_id] = raw_item_id + var item_ids: Array = source_keys.keys() + item_ids.sort() + for item_id: String in item_ids: + contributions.append( + _contribution( + event, location, day_index, item_id, float(ledger[source_keys[item_id]]) + ) + ) + elif payload.has("item_id") and payload.has("amount"): + var raw_item_id: Variant = payload["item_id"] + var raw_amount: Variant = payload["amount"] + if ( + (not raw_item_id is String and not raw_item_id is StringName) + or String(raw_item_id).is_empty() + or (not raw_amount is int and not raw_amount is float) + or not is_finite(float(raw_amount)) + or float(raw_amount) <= 0.0 + ): + return null + contributions.append( + _contribution(event, location, day_index, String(raw_item_id), float(raw_amount)) + ) + else: + return null + return contributions if not contributions.is_empty() else null + + +static func _contribution( + event: WorldEventRecord, + location: SpatialAddress, + day_index: int, + item_id: String, + amount: float +) -> Dictionary: + return { + "day_index": day_index, + "world_id": String(location.get_world_id()), + "location_id": String(location.get_location_id()), + "event_type": String(event.get_event_type()), + "item_id": item_id, + "amount": amount, + } + + +static func _normalize_event_types(values: Array) -> Variant: + var seen: Dictionary = {} + var normalized: Array[StringName] = [] + for value: Variant in values: + if (not value is String and not value is StringName) or StringName(value).is_empty(): + return null + var event_type := StringName(value) + if seen.has(String(event_type)): + return null + seen[String(event_type)] = true + normalized.append(event_type) + normalized.sort_custom( + func(first: StringName, second: StringName) -> bool: return String(first) < String(second) + ) + return normalized + + +static func _normalize_event_ids(values: Array) -> Variant: + var seen: Dictionary = {} + var normalized: Array[int] = [] + for value: Variant in values: + if not value is int or int(value) < 0 or seen.has(int(value)): + return null + seen[int(value)] = true + normalized.append(int(value)) + normalized.sort() + return normalized + + +static func _sorted_int_keys(values: Dictionary) -> Array[int]: + var result: Array[int] = [] + for value: Variant in values: + result.append(int(value)) + result.sort() + return result + + +static func _copy_int_array(values: Array) -> Array[int]: + var result: Array[int] = [] + for value: Variant in values: + result.append(int(value)) + return result + + +static func _copy_string_name_array(values: Array) -> Array[StringName]: + var result: Array[StringName] = [] + for value: Variant in values: + result.append(StringName(value)) + return result + + +static func _reason_count(traces: Array[Dictionary], reason: String) -> int: + var count := 0 + for trace: Dictionary in traces: + var reasons: Array = trace["reasons"] + if reason in reasons: + count += 1 + return count + + +static func _rollup_contribution_count(store: WorldEventDailyRollupStore) -> int: + var count := 0 + for record: WorldEventDailyRollupRecord in store.get_all(): + count += record.get_event_count() + return count diff --git a/simulation/events/WorldEventRetentionPlanner.gd.uid b/simulation/events/WorldEventRetentionPlanner.gd.uid new file mode 100644 index 0000000..d281ed9 --- /dev/null +++ b/simulation/events/WorldEventRetentionPlanner.gd.uid @@ -0,0 +1 @@ +uid://d2s2p1p8556nc diff --git a/simulation/state/WorldEventDailyRollupRecord.gd b/simulation/state/WorldEventDailyRollupRecord.gd new file mode 100644 index 0000000..7864608 --- /dev/null +++ b/simulation/state/WorldEventDailyRollupRecord.gd @@ -0,0 +1,244 @@ +class_name WorldEventDailyRollupRecord +extends RefCounted + +const SCHEMA_VERSION := 1 +const RECORD_FIELDS := [ + "schema_version", + "day_index", + "world_id", + "location_id", + "event_type", + "item_id", + "event_count", + "total_amount", + "minimum_tick", + "maximum_tick", + "minimum_event_id", + "maximum_event_id", + "source_event_ids_checksum", +] + +var _day_index := -1 +var _world_id: StringName +var _location_id: StringName +var _event_type: StringName +var _item_id: StringName +var _event_count := 0 +var _total_amount := 0.0 +var _minimum_tick := -1 +var _maximum_tick := -1 +var _minimum_event_id := -1 +var _maximum_event_id := -1 +var _source_event_ids_checksum := "" + + +static func create( + day_index: int, + world_id: StringName, + location_id: StringName, + event_type: StringName, + item_id: StringName, + source_event_ids: Array[int], + total_amount: float, + minimum_tick: int, + maximum_tick: int +) -> WorldEventDailyRollupRecord: + var normalized_ids: Variant = _normalize_event_ids(source_event_ids) + if normalized_ids == null or normalized_ids.is_empty(): + return null + var record := WorldEventDailyRollupRecord.new() + record._day_index = day_index + record._world_id = world_id + record._location_id = location_id + record._event_type = event_type + record._item_id = item_id + record._event_count = normalized_ids.size() + record._total_amount = total_amount + record._minimum_tick = minimum_tick + record._maximum_tick = maximum_tick + record._minimum_event_id = int(normalized_ids.front()) + record._maximum_event_id = int(normalized_ids.back()) + record._source_event_ids_checksum = JSON.stringify(normalized_ids).sha256_text() + return record if record.is_valid() else null + + +static func from_dictionary(record_data: Dictionary) -> WorldEventDailyRollupRecord: + if not _has_exact_fields(record_data, RECORD_FIELDS): + return null + if ( + not record_data["schema_version"] is int + or int(record_data["schema_version"]) != SCHEMA_VERSION + or not record_data["day_index"] is int + or not record_data["world_id"] is String + or not record_data["location_id"] is String + or not record_data["event_type"] is String + or not record_data["item_id"] is String + or not record_data["event_count"] is int + or (not record_data["total_amount"] is int and not record_data["total_amount"] is float) + or not record_data["minimum_tick"] is int + or not record_data["maximum_tick"] is int + or not record_data["minimum_event_id"] is int + or not record_data["maximum_event_id"] is int + or not record_data["source_event_ids_checksum"] is String + ): + return null + var record := WorldEventDailyRollupRecord.new() + record._day_index = int(record_data["day_index"]) + record._world_id = StringName(record_data["world_id"]) + record._location_id = StringName(record_data["location_id"]) + record._event_type = StringName(record_data["event_type"]) + record._item_id = StringName(record_data["item_id"]) + record._event_count = int(record_data["event_count"]) + record._total_amount = float(record_data["total_amount"]) + record._minimum_tick = int(record_data["minimum_tick"]) + record._maximum_tick = int(record_data["maximum_tick"]) + record._minimum_event_id = int(record_data["minimum_event_id"]) + record._maximum_event_id = int(record_data["maximum_event_id"]) + record._source_event_ids_checksum = String(record_data["source_event_ids_checksum"]) + return record if record.is_valid() else null + + +func is_valid() -> bool: + return ( + _day_index >= 0 + and not _world_id.is_empty() + and not _location_id.is_empty() + and not _event_type.is_empty() + and not _item_id.is_empty() + and _event_count > 0 + and is_finite(_total_amount) + and _total_amount > 0.0 + and _minimum_tick >= 0 + and _maximum_tick >= _minimum_tick + and _minimum_event_id >= 0 + and _maximum_event_id >= _minimum_event_id + and _is_sha256(_source_event_ids_checksum) + ) + + +func get_day_index() -> int: + return _day_index + + +func get_world_id() -> StringName: + return _world_id + + +func get_location_id() -> StringName: + return _location_id + + +func get_event_type() -> StringName: + return _event_type + + +func get_item_id() -> StringName: + return _item_id + + +func get_event_count() -> int: + return _event_count + + +func get_total_amount() -> float: + return _total_amount + + +func get_minimum_tick() -> int: + return _minimum_tick + + +func get_maximum_tick() -> int: + return _maximum_tick + + +func get_minimum_event_id() -> int: + return _minimum_event_id + + +func get_maximum_event_id() -> int: + return _maximum_event_id + + +func get_source_event_ids_checksum() -> String: + return _source_event_ids_checksum + + +func index_key() -> String: + return index_key_for(_day_index, _world_id, _location_id, _event_type, _item_id) + + +func location_index_key() -> String: + return location_key_for(_world_id, _location_id) + + +func get_reason_trace() -> Array[StringName]: + return [ + &"daily_transaction_rollup", + StringName("source_event_count_%d" % _event_count), + StringName("source_event_ids_sha256_%s" % _source_event_ids_checksum), + ] + + +func to_dictionary() -> Dictionary: + return { + "schema_version": SCHEMA_VERSION, + "day_index": _day_index, + "world_id": String(_world_id), + "location_id": String(_location_id), + "event_type": String(_event_type), + "item_id": String(_item_id), + "event_count": _event_count, + "total_amount": _total_amount, + "minimum_tick": _minimum_tick, + "maximum_tick": _maximum_tick, + "minimum_event_id": _minimum_event_id, + "maximum_event_id": _maximum_event_id, + "source_event_ids_checksum": _source_event_ids_checksum, + } + + +static func index_key_for( + day_index: int, + world_id: StringName, + location_id: StringName, + event_type: StringName, + item_id: StringName +) -> String: + return JSON.stringify( + [day_index, String(world_id), String(location_id), String(event_type), String(item_id)] + ) + + +static func location_key_for(world_id: StringName, location_id: StringName) -> String: + return JSON.stringify([String(world_id), String(location_id)]) + + +static func _normalize_event_ids(values: Array) -> Variant: + var seen: Dictionary = {} + var normalized: Array[int] = [] + for value: Variant in values: + if not value is int or int(value) < 0 or seen.has(int(value)): + return null + seen[int(value)] = true + normalized.append(int(value)) + normalized.sort() + return normalized + + +static func _has_exact_fields(record_data: Dictionary, fields: Array) -> bool: + if record_data.size() != fields.size(): + return false + for field: String in fields: + if not record_data.has(field): + return false + return true + + +static func _is_sha256(value: String) -> bool: + if value.length() != 64: + return false + for character_index in range(value.length()): + if value[character_index] not in "0123456789abcdef": + return false + return true diff --git a/simulation/state/WorldEventDailyRollupRecord.gd.uid b/simulation/state/WorldEventDailyRollupRecord.gd.uid new file mode 100644 index 0000000..1744b08 --- /dev/null +++ b/simulation/state/WorldEventDailyRollupRecord.gd.uid @@ -0,0 +1 @@ +uid://b73dsvxt8phkq diff --git a/tests/unit/test_world_event_retention.gd b/tests/unit/test_world_event_retention.gd new file mode 100644 index 0000000..b641595 --- /dev/null +++ b/tests/unit/test_world_event_retention.gd @@ -0,0 +1,299 @@ +extends GutTest + + +func test_daily_rollups_group_by_day_location_item_and_type_deterministically() -> void: + var events: Array[WorldEventRecord] = [ + _transaction(9, &"storage_deposited", 5, &"village_pantry", &"food", 2.0), + _transaction(2, &"storage_deposited", 4, &"village_pantry", &"food", 1.0), + _transaction(4, &"storage_deposited", 8, &"village_pantry", &"wood", 3.0), + _transaction(5, &"storage_deposited", 15, &"village_pantry", &"food", 4.0), + _transaction(6, &"storage_deposited", 15, &"lower_town", &"food", 5.0), + _transaction(7, &"resource_extracted", 15, &"forest_edge", &"food", 6.0), + _transaction(8, &"storage_deposited", 25, &"village_pantry", &"food", 7.0), + ] + var source := _store_from_events(events) + var source_before := source.to_dictionary() + var planner := WorldEventRetentionPlanner.new() + var retention_plan := planner.plan(source, 35, 10, 1, [&"storage_deposited"]) + + assert_not_null(retention_plan) + assert_eq(retention_plan.get_retained_event_ids(), [7, 8]) + assert_eq(retention_plan.get_rolled_up_event_ids(), [2, 4, 5, 6, 9]) + assert_eq(source.to_dictionary(), source_before) + var rollups := retention_plan.get_rollup_store() + assert_eq(rollups.size(), 4) + var day_zero_food := rollups.get_by_key( + 0, &"world_core", &"village_pantry", &"storage_deposited", &"food" + ) + assert_not_null(day_zero_food) + assert_eq(day_zero_food.get_event_count(), 2) + assert_eq(day_zero_food.get_total_amount(), 3.0) + assert_eq(day_zero_food.get_minimum_tick(), 4) + assert_eq(day_zero_food.get_maximum_tick(), 5) + assert_eq(day_zero_food.get_minimum_event_id(), 2) + assert_eq(day_zero_food.get_maximum_event_id(), 9) + assert_eq(day_zero_food.get_source_event_ids_checksum(), JSON.stringify([2, 9]).sha256_text()) + assert_eq(rollups.get_for_day(0).size(), 2) + assert_eq(rollups.get_for_location(&"world_core", &"village_pantry").size(), 3) + assert_eq(rollups.get_for_item(&"food").size(), 3) + assert_eq(rollups.get_for_event_type(&"storage_deposited").size(), 4) + assert_eq(retention_plan.get_reason_trace(7)["reasons"], ["not_rollup_eligible"]) + assert_eq(retention_plan.get_reason_trace(8)["reasons"], ["within_raw_retention"]) + var telemetry := retention_plan.get_telemetry() + assert_eq(telemetry["input_event_count"], 7) + assert_eq(telemetry["retained_event_count"], 2) + assert_eq(telemetry["rolled_up_event_count"], 5) + assert_eq(telemetry["rollup_record_count"], 4) + assert_eq(telemetry["rollup_contribution_count"], 5) + assert_true(telemetry["source_unchanged"]) + assert_eq(planner.get_last_telemetry(), telemetry) + + var reversed_events := events.duplicate() + reversed_events.reverse() + var reversed_plan := WorldEventRetentionPlanner.new().plan( + _store_from_events(reversed_events), 35, 10, 1, [&"storage_deposited"] + ) + assert_not_null(reversed_plan) + assert_eq(reversed_plan.to_dictionary(), retention_plan.to_dictionary()) + assert_eq(reversed_plan.checksum(), retention_plan.checksum()) + + +func test_pinned_external_and_payload_causal_facts_remain_exact_raw_events() -> void: + var source := WorldEventStore.new() + assert_true(source.append(_causal_transaction(1, 1, -1))) + assert_true(source.append(_causal_transaction(2, 2, 1))) + assert_true( + source.append( + WorldEventRecord.create( + 3, &"situation_opened", 3, {}, _address(&"village_pantry"), {"cause_event_id": 2} + ) + ) + ) + assert_true(source.append(_causal_transaction(4, 4, -1))) + assert_true(source.append(_causal_transaction(5, 5, -1))) + assert_true(source.append(_causal_transaction(6, 6, -1))) + var planner := WorldEventRetentionPlanner.new() + var retention_plan := planner.plan(source, 50, 10, 0, [&"storage_deposited"], [4], [5]) + + assert_not_null(retention_plan) + assert_eq(retention_plan.get_retained_event_ids(), [1, 2, 3, 4, 5]) + assert_eq(retention_plan.get_rolled_up_event_ids(), [6]) + assert_has(retention_plan.get_reason_trace(1)["reasons"], "causal_ancestor") + assert_eq(retention_plan.get_reason_trace(1)["required_by_event_ids"], [2]) + assert_has(retention_plan.get_reason_trace(2)["reasons"], "causal_ancestor") + assert_eq(retention_plan.get_reason_trace(2)["required_by_event_ids"], [3]) + assert_has(retention_plan.get_reason_trace(3)["reasons"], "not_rollup_eligible") + assert_has(retention_plan.get_reason_trace(4)["reasons"], "pinned_fact") + assert_has(retention_plan.get_reason_trace(5)["reasons"], "explicit_causal_fact") + assert_eq(retention_plan.get_reason_trace(6)["decision"], "rolled_up") + assert_eq(retention_plan.get_telemetry()["causal_ancestor_count"], 2) + var missing_causal_trace := retention_plan.to_dictionary() + missing_causal_trace["reason_traces"][0]["reasons"].erase("causal_ancestor") + assert_null(WorldEventRetentionPlan.from_dictionary(missing_causal_trace)) + var missing_pin_trace := retention_plan.to_dictionary() + missing_pin_trace["reason_traces"][3]["reasons"].erase("pinned_fact") + assert_null(WorldEventRetentionPlan.from_dictionary(missing_pin_trace)) + + var before_missing_pin := source.to_dictionary() + assert_null(planner.plan(source, 50, 10, 0, [&"storage_deposited"], [404])) + assert_string_contains(planner.get_last_error(), "pinned fact ID 404 is missing") + assert_false(planner.get_last_telemetry()["succeeded"]) + assert_eq(source.to_dictionary(), before_missing_pin) + + +func test_cargo_ledger_rolls_up_each_item_and_invalid_transactions_stay_raw() -> void: + var source := WorldEventStore.new() + assert_true( + source.append( + WorldEventRecord.create( + 10, + &"caravan_cargo_deposited", + 1, + {}, + _address(&"river_port"), + {"cargo_ledger": {"wood": 1.0, "food": 2.5}, "cause_event_id": -1} + ) + ) + ) + assert_true( + source.append( + WorldEventRecord.create( + 11, + &"caravan_cargo_deposited", + 2, + {}, + _address(&"river_port"), + {"cargo_ledger": {"food": 1.0}, "item_id": "food", "amount": 1.0} + ) + ) + ) + assert_true( + source.append( + WorldEventRecord.create( + 12, + &"caravan_cargo_deposited", + 2, + {}, + _address(&"river_port"), + {"cargo_ledger": {"food": 0.0}} + ) + ) + ) + assert_true( + source.append( + WorldEventRecord.create( + 13, &"caravan_cargo_deposited", 2, {}, null, {"cargo_ledger": {"food": 1.0}} + ) + ) + ) + var retention_plan := WorldEventRetentionPlanner.new().plan( + source, 30, 10, 0, [&"caravan_cargo_deposited"] + ) + + assert_not_null(retention_plan) + assert_eq(retention_plan.get_rolled_up_event_ids(), [10]) + assert_eq(retention_plan.get_retained_event_ids(), [11, 12, 13]) + var rollups := retention_plan.get_rollup_store() + assert_eq(rollups.size(), 2) + assert_eq( + ( + rollups + . get_by_key(0, &"world_core", &"river_port", &"caravan_cargo_deposited", &"food") + . get_total_amount() + ), + 2.5 + ) + assert_eq( + ( + rollups + . get_by_key(0, &"world_core", &"river_port", &"caravan_cargo_deposited", &"wood") + . get_total_amount() + ), + 1.0 + ) + assert_eq(retention_plan.get_reason_trace(10)["rollup_keys"].size(), 2) + for event_id in [11, 12, 13]: + assert_has(retention_plan.get_reason_trace(event_id)["reasons"], "rollup_payload_invalid") + assert_eq(retention_plan.get_telemetry()["invalid_rollup_event_count"], 3) + assert_eq(retention_plan.get_telemetry()["rollup_contribution_count"], 2) + + +func test_record_store_and_plan_are_canonical_primitive_and_round_trippable() -> void: + var source := _store_from_events( + [ + _transaction(3, &"storage_deposited", 1, &"pantry", &"food", 2.0), + _transaction(4, &"storage_deposited", 2, &"pantry", &"food", 3.0), + ] + ) + var retention_plan := WorldEventRetentionPlanner.new().plan( + source, 20, 10, 0, [&"storage_deposited"] + ) + var serialized := retention_plan.to_dictionary() + var restored := WorldEventRetentionPlan.from_dictionary(serialized) + + assert_not_null(restored) + assert_true(_is_primitive_tree(serialized)) + assert_eq(restored.to_dictionary(), serialized) + assert_eq(JSON.stringify(restored.to_dictionary()), JSON.stringify(serialized)) + assert_eq(restored.checksum(), retention_plan.checksum()) + var rollup := restored.get_rollup_store().get_all()[0] + assert_has(rollup.get_reason_trace(), &"daily_transaction_rollup") + assert_true(String(rollup.get_reason_trace()[-1]).begins_with("source_event_ids_sha256_")) + assert_eq( + WorldEventDailyRollupRecord.from_dictionary(rollup.to_dictionary()).to_dictionary(), + rollup.to_dictionary() + ) + + var store := restored.get_rollup_store() + var store_before := store.to_dictionary() + var duplicated_store := store_before.duplicate(true) + duplicated_store["rollups"].append(store_before["rollups"][0].duplicate(true)) + assert_false(store.restore_from_dictionary(duplicated_store)) + assert_eq(store.to_dictionary(), store_before) + var noncanonical_plan := serialized.duplicate(true) + noncanonical_plan["reason_traces"].reverse() + assert_null(WorldEventRetentionPlan.from_dictionary(noncanonical_plan)) + var invalid_record := rollup.to_dictionary() + invalid_record["total_amount"] = -1.0 + assert_null(WorldEventDailyRollupRecord.from_dictionary(invalid_record)) + var false_source_checksum := serialized.duplicate(true) + false_source_checksum["rollup_store"]["rollups"][0]["source_event_ids_checksum"] = ( + JSON.stringify([3]).sha256_text() + ) + assert_null(WorldEventRetentionPlan.from_dictionary(false_source_checksum)) + + +func test_planner_rejects_dangling_and_non_prior_causal_facts_without_mutation() -> void: + var dangling := WorldEventStore.new() + assert_true(dangling.append(_causal_transaction(1, 1, 99))) + var dangling_before := dangling.to_dictionary() + var planner := WorldEventRetentionPlanner.new() + assert_null(planner.plan(dangling, 20, 10, 0, [&"storage_deposited"] as Array)) + assert_string_contains(planner.get_last_error(), "missing causal fact 99") + assert_eq(dangling.to_dictionary(), dangling_before) + + var forward := WorldEventStore.new() + assert_true(forward.append(_causal_transaction(1, 5, 2))) + assert_true(forward.append(_causal_transaction(2, 5, -1))) + var forward_before := forward.to_dictionary() + assert_null(planner.plan(forward, 20, 10, 0, [&"storage_deposited"] as Array)) + assert_string_contains(planner.get_last_error(), "non-prior causal fact 2") + assert_eq(forward.to_dictionary(), forward_before) + + +func _transaction( + event_id: int, + event_type: StringName, + tick: int, + location_id: StringName, + item_id: StringName, + amount: float +) -> WorldEventRecord: + return WorldEventRecord.create( + event_id, + event_type, + tick, + {}, + _address(location_id), + {"item_id": String(item_id), "amount": amount} + ) + + +func _causal_transaction(event_id: int, tick: int, cause_event_id: int) -> WorldEventRecord: + return WorldEventRecord.create( + event_id, + &"storage_deposited", + tick, + {}, + _address(&"village_pantry"), + {"item_id": "food", "amount": 1.0, "cause_event_id": cause_event_id} + ) + + +func _address(location_id: StringName) -> SpatialAddress: + return SpatialAddress.create(&"world_core", location_id, Vector3.ZERO) + + +func _store_from_events(events: Array[WorldEventRecord]) -> WorldEventStore: + var store := WorldEventStore.new() + for event: WorldEventRecord in events: + assert_true(store.append(event)) + return store + + +func _is_primitive_tree(value: Variant) -> bool: + if value == null or value is bool or value is int or value is String: + return true + if value is float: + return is_finite(value) + if value is Array: + for item: Variant in value: + if not _is_primitive_tree(item): + return false + return true + if value is Dictionary: + for key: Variant in value: + if not key is String or not _is_primitive_tree(value[key]): + return false + return true + return false diff --git a/tests/unit/test_world_event_retention.gd.uid b/tests/unit/test_world_event_retention.gd.uid new file mode 100644 index 0000000..be6ca4e --- /dev/null +++ b/tests/unit/test_world_event_retention.gd.uid @@ -0,0 +1 @@ +uid://d3siqr8dnivfn