engine/modules/agent_events/agent_events.cpp

#include "agent_events.h"

#include "core/io/json.h"
#include "core/os/os.h"
#include "modules/agent_api/agent_server.h"

AgentEvents *AgentEvents::singleton = nullptr;

AgentEvents::AgentEvents() {
	singleton = this;
}

AgentEvents::~AgentEvents() {
	singleton = nullptr;
}

void AgentEvents::_bind_methods() {
	ClassDB::bind_method(D_METHOD("emit_event", "type", "data", "source"), &AgentEvents::emit_event, DEFVAL(Dictionary()), DEFVAL("game"));
	ClassDB::bind_method(D_METHOD("on_node_added", "path", "class_name"), &AgentEvents::on_node_added);
	ClassDB::bind_method(D_METHOD("on_node_removed", "path", "class_name"), &AgentEvents::on_node_removed);
	ClassDB::bind_method(D_METHOD("on_collision", "body_a", "body_b", "data"), &AgentEvents::on_collision, DEFVAL(Dictionary()));
	ClassDB::bind_method(D_METHOD("on_input_action", "action", "pressed", "strength"), &AgentEvents::on_input_action);
	ClassDB::bind_method(D_METHOD("on_animation_event", "node", "anim", "event_type"), &AgentEvents::on_animation_event);
	ClassDB::bind_method(D_METHOD("get_history", "count", "type", "since_msec"), &AgentEvents::get_history, DEFVAL(100), DEFVAL(""), DEFVAL(0));
	ClassDB::bind_method(D_METHOD("get_event_count"), &AgentEvents::get_event_count);
	ClassDB::bind_method(D_METHOD("subscribe", "type"), &AgentEvents::subscribe);
	ClassDB::bind_method(D_METHOD("unsubscribe", "type"), &AgentEvents::unsubscribe);
	ClassDB::bind_method(D_METHOD("set_subscribe_all", "all"), &AgentEvents::set_subscribe_all);
	ClassDB::bind_method(D_METHOD("set_max_events_per_sec", "max"), &AgentEvents::set_max_events_per_sec);
	ClassDB::bind_method(D_METHOD("clear"), &AgentEvents::clear);
}

void AgentEvents::emit_event(const String &p_type, const Dictionary &p_data, const String &p_source) {
	Event evt;
	evt.timestamp_msec = OS::get_singleton()->get_ticks_msec();
	evt.event_type = p_type;
	evt.source = p_source;
	evt.data = p_data;

	_push_event(evt);
}

void AgentEvents::on_node_added(const String &p_path, const String &p_class) {
	Dictionary data;
	data["path"] = p_path;
	data["class"] = p_class;
	emit_event("node_added", data, "engine");
}

void AgentEvents::on_node_removed(const String &p_path, const String &p_class) {
	Dictionary data;
	data["path"] = p_path;
	data["class"] = p_class;
	emit_event("node_removed", data, "engine");
}

void AgentEvents::on_collision(const String &p_body_a, const String &p_body_b, const Dictionary &p_data) {
	Dictionary data = p_data;
	data["body_a"] = p_body_a;
	data["body_b"] = p_body_b;
	emit_event("collision", data, "physics");
}

void AgentEvents::on_input_action(const String &p_action, bool p_pressed, float p_strength) {
	Dictionary data;
	data["action"] = p_action;
	data["pressed"] = p_pressed;
	data["strength"] = p_strength;
	emit_event("input_action", data, "input");
}

void AgentEvents::on_animation_event(const String &p_node, const String &p_anim, const String &p_event_type) {
	Dictionary data;
	data["node"] = p_node;
	data["animation"] = p_anim;
	data["event_type"] = p_event_type;
	emit_event("animation", data, "animation");
}

void AgentEvents::_push_event(const Event &p_event) {
	// Rate limiting.
	uint64_t now = OS::get_singleton()->get_ticks_msec();
	if (last_emit_time.has(p_event.event_type)) {
		uint64_t delta = now - last_emit_time[p_event.event_type];
		if (delta < (uint64_t)(1000 / max_events_per_sec)) {
			return; // Rate limited.
		}
	}
	last_emit_time[p_event.event_type] = now;

	Event evt = p_event;

	{
		MutexLock lock(history_mutex);
		evt.id = next_id++;
		history[write_pos] = evt;
		write_pos = (write_pos + 1) % HISTORY_SIZE;
		if (event_count < HISTORY_SIZE) {
			event_count++;
		}
	}

	// Push to SSE via AgentServer.
	if (!subscribe_all && !subscribed_types.has(evt.event_type)) {
		return;
	}

	AgentServer *server = AgentServer::get_singleton();
	if (server && server->is_running()) {
		Dictionary dict;
		dict["id"] = evt.id;
		dict["timestamp_msec"] = evt.timestamp_msec;
		dict["type"] = evt.event_type;
		dict["source"] = evt.source;
		dict["data"] = evt.data;
		server->push_sse("events", evt.event_type, JSON::stringify(dict));
	}
}

Array AgentEvents::get_history(int p_count, const String &p_type, uint64_t p_since_msec) {
	Array result;
	MutexLock lock(history_mutex);

	int start = (event_count < HISTORY_SIZE) ? 0 : write_pos;
	int count = event_count;

	for (int i = count - 1; i >= 0 && result.size() < p_count; i--) {
		int idx = (start + i) % HISTORY_SIZE;
		const Event &evt = history[idx];

		if (!p_type.is_empty() && evt.event_type != p_type) {
			continue;
		}
		if (p_since_msec > 0 && evt.timestamp_msec < p_since_msec) {
			break;
		}

		Dictionary dict;
		dict["id"] = evt.id;
		dict["timestamp_msec"] = evt.timestamp_msec;
		dict["type"] = evt.event_type;
		dict["source"] = evt.source;
		dict["data"] = evt.data;
		result.push_back(dict);
	}

	return result;
}

void AgentEvents::subscribe(const String &p_type) {
	subscribed_types[p_type] = true;
}

void AgentEvents::unsubscribe(const String &p_type) {
	subscribed_types.erase(p_type);
}

void AgentEvents::clear() {
	MutexLock lock(history_mutex);
	write_pos = 0;
	event_count = 0;
}