clanker-saddle/src/Services/EventService.vala

using Invercargill;
using Invercargill.DataStructures;
using Invercargill.Expressions;
using InvercargillSql.Orm;
using Inversion;
using ClankerSaddle.Data;

namespace ClankerSaddle.Services {

    /** Event kinds (the `events.kind` values, plan 17 §3.1/§7.2). */
    public const string EVENT_TURN_TEXT = "turn_text";
    public const string EVENT_TOOL_CALL = "tool_call";
    public const string EVENT_TOOL_RESULT = "tool_result";
    public const string EVENT_COMMAND_OUTPUT = "command_output";
    public const string EVENT_PROGRESS = "progress";
    public const string EVENT_COMMENT = "comment";
    public const string EVENT_STATE_CHANGE = "state_change";
    public const string EVENT_LIFECYCLE = "lifecycle";
    public const string EVENT_PLAN_SUBMITTED = "plan_submitted";
    public const string EVENT_VERDICT = "verdict";
    public const string EVENT_FORCE_WAKE = "force_wake";
    public const string EVENT_USAGE = "usage";
    public const string EVENT_ANSWER = "answer";
    public const string EVENT_CANCEL = "cancel";

    /** Event actors (the `events.actor` values). */
    public const string EVENT_ACTOR_MODEL = "model";
    public const string EVENT_ACTOR_HARNESS = "harness";
    public const string EVENT_ACTOR_USER = "user";
    public const string EVENT_ACTOR_SYSTEM = "system";

    /** Event scopes (the `events.scope_type` values). */
    public const string EVENT_SCOPE_PROJECT = "project";
    public const string EVENT_SCOPE_GOAL = "goal";
    public const string EVENT_SCOPE_TASK = "task";
    public const string EVENT_SCOPE_SESSION = "session";

    /** The event log page/SSE window (plan 17 §3.2). */
    public const int EVENT_PAGE_SIZE = 30;

    /**
     * The plan-17 unified event ledger's writer and read model: daemon
     * event ingestion from `POST /api/saddle/chunks` (idempotent on
     * UNIQUE(session_id, seq)), web-originated appends from the domain
     * mutations, and the paged/newer-than queries the event log page,
     * the timeline, the goal narrative and the resume machinery read.
     */
    public class EventService : Object {

        private OrmSession db = inject<OrmSession>();

        // ------------------------------------------------------------------
        // Writers
        // ------------------------------------------------------------------

        /**
         * Appends one web-originated event: session_id 0 and seq NULL
         * (the unique constraint never bites — SQLite treats NULLs as
         * distinct), `created_at` stamped now. `payload` may be null
         * (an empty object stores).
         */
        public async EventRecord append(string kind, string actor, string scope_type,
                int64 scope_id, Json.Object? payload) throws Error {
            var record = new EventRecord() {
                kind = kind,
                actor = actor,
                scope_type = scope_type,
                scope_id = scope_id,
                payload = payload != null ? json_to_text((!)payload) : "{}",
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<EventRecord>(record);
            return record;
        }

        /**
         * Stores a pushed batch of daemon events (plan 17 §7.2: the
         * `events[]` of `POST /api/saddle/chunks`) as ledger rows
         * scoped to the session's task. Idempotent on the unique
         * (session_id, seq) pair: a redelivered seq is consumed, not
         * an error; an event without a positive seq is refused (the
         * daemon contract makes seq mandatory). The daemon's `ts` is
         * authoritative for `created_at` (fallback: now). Returns the
         * count of rows newly stored.
         */
        public async int64 store_pushed_events(int64 session_id, int64 task_id,
                Json.Array events) throws Error {
            int64 stored = 0;
            var staged = new Vector<EventRecord>();
            foreach (var element in events.get_elements()) {
                if (element.get_node_type() != Json.NodeType.OBJECT) {
                    continue;
                }
                var incoming = element.get_object();
                int64 seq = incoming.has_member("seq")
                    && incoming.get_member("seq").get_value_type() == typeof(int64)
                    ? incoming.get_int_member("seq") : 0;
                if (seq <= 0) {
                    throw new Error(Quark.from_string("clanker-saddle"), 90,
                        "Every pushed event needs a positive per-session seq");
                }
                string kind = EntryJson.string_member(incoming, "kind");
                if (kind.length == 0) {
                    throw new Error(Quark.from_string("clanker-saddle"), 91,
                        "Every pushed event needs a kind");
                }
                string actor = EntryJson.string_member(incoming, "actor");
                var payload_node = incoming.has_member("payload")
                    && incoming.get_member("payload").get_node_type() == Json.NodeType.OBJECT
                    ? incoming.get_member("payload") : null;
                staged.add(new EventRecord() {
                    session_id = session_id,
                    seq = seq,
                    kind = kind,
                    actor = actor.length > 0 ? actor : EVENT_ACTOR_HARNESS,
                    scope_type = EVENT_SCOPE_TASK,
                    scope_id = task_id,
                    payload = payload_node != null
                        ? json_to_text((!)payload_node.get_object()) : "{}",
                    created_at = EntryJson.parse_iso(incoming, "ts")
                        ?? new DateTime.now_local()
                });
            }
            foreach (var record in staged) {
                try {
                    yield db.insert_async<EventRecord>(record);
                    stored++;
                } catch (Error insert_error) {
                    var existing = yield db.query<EventRecord>()
                        .where(expr("session_id == $0 && seq == $1",
                            elem<int64?>(session_id), elem<int64?>(record.seq)))
                        .first_async();
                    if (existing == null) {
                        throw insert_error;
                    }
                }
            }
            return stored;
        }

        // ------------------------------------------------------------------
        // Readers
        // ------------------------------------------------------------------

        /**
         * One newest-first page of a task's ledger (plan 17 §7.4):
         * events with id below `before` (0 = from the top), `limit`
         * rows, newest first.
         */
        public async ImmutableLot<EventRecord> page_for_task(int64 task_id,
                int64 before, int64 limit) throws Error {
            var query = db.query<EventRecord>()
                .where(expr("scope_type == $0 && scope_id == $1",
                    elem(EVENT_SCOPE_TASK), elem<int64?>(task_id)));
            if (before > 0) {
                query.where(expr("id < $0", elem<int64?>(before)));
            }
            return yield query
                .order_by_desc(expr("id"))
                .limit(limit)
                .materialise_async();
        }

        /**
         * The task's events strictly newer than `after_id`, oldest
         * first — the SSE stream's live-append cursor walk.
         */
        public async ImmutableLot<EventRecord> newer_than(int64 task_id, int64 after_id)
                throws Error {
            return yield db.query<EventRecord>()
                .where(expr("scope_type == $0 && scope_id == $1 && id > $2",
                    elem(EVENT_SCOPE_TASK), elem<int64?>(task_id), elem<int64?>(after_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /**
         * The task's events of the given kinds, newest first — the
         * progress feed and the goal narrative's filtered projections.
         */
        public async ImmutableLot<EventRecord> list_for_task(int64 task_id,
                string[] kinds) throws Error {
            var query = db.query<EventRecord>()
                .where(expr("scope_type == $0 && scope_id == $1",
                    elem(EVENT_SCOPE_TASK), elem<int64?>(task_id)));
            if (kinds.length > 0) {
                query.where(kind_in(kinds));
            }
            return yield query
                .order_by_desc(expr("id"))
                .materialise_async();
        }

        /** The goal's own events (scope goal), newest first. */
        public async ImmutableLot<EventRecord> list_for_goal(int64 goal_id) throws Error {
            return yield db.query<EventRecord>()
                .where(expr("scope_type == $0 && scope_id == $1",
                    elem(EVENT_SCOPE_GOAL), elem<int64?>(goal_id)))
                .order_by_desc(expr("id"))
                .materialise_async();
        }

        /**
         * The newest terminal state_change (to done/failed) among the
         * tasks' events that is newer than `watermark` — the resume
         * coalescer's oldest-pending-transition finder. Null when no
         * unconsumed terminal transition exists.
         */
        public async DateTime? oldest_terminal_transition_since(
                Invercargill.DataStructures.HashSet<string> task_ids,
                DateTime? watermark) throws Error {
            var filter = TaskService.id_in("scope_id", task_ids);
            if (filter == null) {
                return null;
            }
            DateTime? oldest = null;
            foreach (var record in yield db.query<EventRecord>()
                    .where(expr("kind == $0 && scope_type == $1",
                        elem(EVENT_STATE_CHANGE), elem(EVENT_SCOPE_TASK)))
                    .where((!)filter)
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (!is_terminal_transition(record)) {
                    continue;
                }
                if (record.created_at == null) {
                    continue;
                }
                if (watermark != null
                        && ((!)record.created_at).compare((!)watermark) <= 0) {
                    continue;
                }
                if (oldest == null || ((!)record.created_at).compare((!)oldest) < 0) {
                    oldest = record.created_at;
                }
            }
            return oldest;
        }

        /**
         * The newest created_at among the tasks' terminal state_change
         * events — the watermark advance (null when none exist).
         */
        public async DateTime? newest_terminal_transition(
                Invercargill.DataStructures.HashSet<string> task_ids) throws Error {
            var filter = TaskService.id_in("scope_id", task_ids);
            if (filter == null) {
                return null;
            }
            DateTime? newest = null;
            foreach (var record in yield db.query<EventRecord>()
                    .where(expr("kind == $0 && scope_type == $1",
                        elem(EVENT_STATE_CHANGE), elem(EVENT_SCOPE_TASK)))
                    .where((!)filter)
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (!is_terminal_transition(record) || record.created_at == null) {
                    continue;
                }
                if (newest == null || ((!)record.created_at).compare((!)newest) > 0) {
                    newest = record.created_at;
                }
            }
            return newest;
        }

        /** True when the record is a state_change to done or failed. */
        private static bool is_terminal_transition(EventRecord record) {
            if (record.kind != EVENT_STATE_CHANGE) {
                return false;
            }
            string to = payload_string(record, "to");
            return to == STATUS_DONE || to == STATUS_FAILED;
        }

        /**
         * The task's newest event of the "what is it doing" kinds —
         * the goal page's current-activity line. Null when none.
         */
        public async EventRecord? newest_activity(int64 task_id) throws Error {
            EventRecord? newest = null;
            foreach (var record in yield list_for_task(task_id,
                    { EVENT_TURN_TEXT, EVENT_TOOL_CALL, EVENT_TOOL_RESULT,
                      EVENT_PROGRESS, EVENT_LIFECYCLE, EVENT_COMMENT })) {
                newest = record;
                break;
            }
            return newest;
        }

        /** The newest event stamp of one session (any kind), or null. */
        public async DateTime? last_session_activity(int64 task_id, int64 session_id)
                throws Error {
            DateTime? newest = null;
            foreach (var record in yield db.query<EventRecord>()
                    .where(expr("scope_type == $0 && scope_id == $1 && session_id == $2",
                        elem(EVENT_SCOPE_TASK), elem<int64?>(task_id),
                        elem<int64?>(session_id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (record.created_at != null) {
                    newest = record.created_at;
                }
            }
            return newest;
        }

        /** Deletes every event scoped to the task (the delete cascade). */
        public async void delete_for_task(int64 task_id) throws Error {
            foreach (var record in yield db.query<EventRecord>()
                    .where(expr("scope_type == $0 && scope_id == $1",
                        elem(EVENT_SCOPE_TASK), elem<int64?>(task_id)))
                    .materialise_async()) {
                yield db.delete_async<EventRecord>(record);
            }
        }

        // ------------------------------------------------------------------
        // Payload helpers
        // ------------------------------------------------------------------

        /** The record's payload as a JSON object (empty when malformed). */
        public static Json.Object payload_of(EventRecord record) {
            try {
                var node = Json.from_string(record.payload ?? "{}");
                if (node != null && node.get_node_type() == Json.NodeType.OBJECT) {
                    return node.get_object();
                }
            } catch (Error e) {
            }
            return new Json.Object();
        }

        /** One string member of the record's payload ("" when absent). */
        public static string payload_string(EventRecord record, string member) {
            return EntryJson.string_member(payload_of(record), member);
        }

        /** One string member of the record's payload (null when absent). */
        public static string? payload_optional_string(EventRecord record, string member) {
            return EntryJson.optional_string_member(payload_of(record), member);
        }

        /** The record's payload serialised back to compact JSON text. */
        public static string json_to_text(Json.Object payload) {
            var generator = new Json.Generator();
            generator.set_root(new Json.Node(Json.NodeType.OBJECT).init_object(payload));
            return generator.to_data(null);
        }

        /** An OR-chain over the kind column for a batch of kinds. */
        private static Expression kind_in(string[] kinds) throws Error {
            Expression? chain = null;
            foreach (var kind in kinds) {
                var comparison = expr("kind == $0", elem(kind));
                chain = chain == null
                    ? comparison : new BinaryExpression((!)chain, comparison,
                        BinaryOperator.OR);
            }
            return (!)chain;
        }
    }
}