;; Title: Miniforge.ai
;; Subtitle: An agentic SDLC / fleet-control platform
;; Author: Christopher Lester
;; Line: Founder, Miniforge.ai (project)
;; Copyright 2025-2026 Christopher Lester (christopher@miniforge.ai)
;;
;; Licensed under the Apache License, Version 2.0 (the "License");
;; you may not use this file except in compliance with the License.
;; You may obtain a copy of the License at
;;
;;     http://www.apache.org/licenses/LICENSE-2.0
;;
;; Unless required by applicable law or agreed to in writing, software
;; distributed under the License is distributed on an "AS IS" BASIS,
;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;; See the License for the specific language governing permissions and
;; limitations under the License.

(ns ai.miniforge.supervisory-state.core-test
  "End-to-end tests for the supervisory-state component lifecycle:
   replay, live subscription, snapshot emission, and re-emit prevention."
  (:require
   [clojure.test :refer [deftest is]]
   [ai.miniforge.event-stream.core :as es-core]
   [ai.miniforge.event-stream.interface :as es]
   [ai.miniforge.supervisory-state.core :as core]
   [ai.miniforge.supervisory-state.interface :as iface]))

;------------------------------------------------------------------------------ Helpers

(defn- no-sink-stream []
  (es-core/create-event-stream {:sinks []}))

(defn- supervisory-events
  "Events on the stream with type in the `:supervisory/*` family."
  [stream]
  (->> (es/get-events stream)
       (filter #(some-> % :event/type namespace (= "supervisory")))))

(defn- workflow-started [wf-id]
  {:event/type :workflow/started
   :event/id (random-uuid)
   :event/timestamp (java.util.Date.)
   :event/version "1.0.0"
   :event/sequence-number 0
   :workflow/id wf-id
   :message "Workflow started"})

(defn- workflow-completed [wf-id]
  {:event/type :workflow/completed
   :event/id (random-uuid)
   :event/timestamp (java.util.Date.)
   :event/version "1.0.0"
   :event/sequence-number 1
   :workflow/id wf-id
   :message "Workflow completed"})

;------------------------------------------------------------------------------ Lifecycle

(deftest start-then-stop-subscribes-and-unsubscribes
  (let [stream (no-sink-stream)
        comp   (iface/create stream)]
    (iface/start! comp)
    (is (true? (:subscribed? @comp)))
    (is (contains? (:subscribers @stream) core/subscriber-id))
    (iface/stop! comp)
    (is (false? (:subscribed? @comp)))
    (is (not (contains? (:subscribers @stream) core/subscriber-id)))))

(deftest attach!-is-create-plus-start!
  (let [stream (no-sink-stream)
        comp   (iface/attach! stream)]
    (is (true? (:subscribed? @comp))
        "attach! returns a component that is already subscribed")
    (is (contains? (:subscribers @stream) core/subscriber-id)
        "attach! registers the stream subscription identically to create+start!")
    (iface/stop! comp)))

(deftest ensure-attached!-is-idempotent
  (let [stream (no-sink-stream)]
    (is (false? (iface/attached? stream)))
    (let [created (iface/ensure-attached! stream)]
      (is (some? created))
      (is (true? (iface/attached? stream)))
      (is (= 1 (count (:subscribers @stream)))))
    (is (nil? (iface/ensure-attached! stream)))
    (is (= 1 (count (:subscribers @stream))))))

(deftest ensure-attached!-is-atomic-per-stream
  (let [stream (no-sink-stream)
        attempts 8
        results (->> (range attempts)
                     (mapv (fn [_]
                             (future (iface/ensure-attached! stream))))
                     (mapv deref))
        created-count (count (filter some? results))]
    (is (= 1 created-count))
    (is (true? (iface/attached? stream)))
    (is (= 1 (count (:subscribers @stream))))))

;------------------------------------------------------------------------------ Emission

(deftest live-workflow-events-produce-supervisory-snapshots
  (let [stream (no-sink-stream)
        comp   (iface/create stream)
        wf-id  (random-uuid)]
    (iface/start! comp)
    (es/publish! stream (workflow-started wf-id))
    (es/publish! stream (workflow-completed wf-id))
    (let [snaps   (supervisory-events stream)
          kinds   (frequencies (map :event/type snaps))]
      (is (>= (count snaps) 2)
          "at least one supervisory/workflow-upserted per entity change")
      (is (some #{:supervisory/workflow-upserted} (keys kinds)))
      ;; attention may also fire (workflow completed → :info item)
      )))

(deftest supervisory-events-are-not-re-emitted
  (let [stream (no-sink-stream)
        comp   (iface/create stream)]
    (iface/start! comp)
    (es/publish! stream (workflow-started (random-uuid)))
    (let [before (count (supervisory-events stream))]
      ;; Publish a synthetic supervisory event directly; handle-event! must
      ;; ignore it so the component doesn't recurse.
      (es/publish! stream {:event/type :supervisory/workflow-upserted
                           :event/id (random-uuid)
                           :event/timestamp (java.util.Date.)
                           :event/version "1.0.0"
                           :event/sequence-number 99
                           :workflow/id (random-uuid)
                           :message "synthetic"
                           :supervisory/entity {:workflow-run/id (random-uuid)}})
      (let [after (count (supervisory-events stream))]
        (is (= (inc before) after)
            "only the synthetic event should be added; no re-emission loop")))))

;------------------------------------------------------------------------------ Live view build-up

(deftest live-events-build-the-view-incrementally
  ;; Covers the production attach order: component attaches first, then
  ;; events flow. Replaces the former startup-replay test after YAGNI-
  ;; removing `replay` (no production caller ever populated the stream
  ;; before the component subscribed).
  (let [stream (no-sink-stream)
        wf-id  (random-uuid)
        comp   (iface/start! (iface/create stream))]
    (es/publish! stream (workflow-started wf-id))
    (es/publish! stream (workflow-completed wf-id))
    (let [runs (iface/workflows comp)]
      (is (= 1 (count runs)))
      (is (= :completed (:workflow-run/status (first runs)))
          "live subscription must see all events, ending in :completed"))))

;------------------------------------------------------------------------------ TaskNode (N5-δ3 §3.3)

(defn- task-state-changed [tid wf-id to-state & [context]]
  (cond-> {:event/type :task/state-changed
           :event/id (random-uuid)
           :event/timestamp (java.util.Date.)
           :event/version "1.0.0"
           :event/sequence-number 0
           :workflow/id wf-id
           :task/id tid
           :task/to-state to-state
           :message (str "Task " tid " → " (name to-state))}
    context (assoc :task/context context)))

(deftest task-state-changed-produces-supervisory-task-node-upserted
  (let [stream (no-sink-stream)
        comp   (iface/create stream)
        tid    (random-uuid)
        wf-id  (random-uuid)]
    (iface/start! comp)
    (es/publish! stream (task-state-changed tid wf-id :running
                                            {:description "Implement X" :type :implement}))
    (let [snaps (->> (supervisory-events stream)
                     (filter #(= :supervisory/task-node-upserted (:event/type %))))]
      (is (= 1 (count snaps)))
      (let [entity (:supervisory/entity (first snaps))]
        (is (= tid (:task/id entity)))
        (is (= wf-id (:task/workflow-run-id entity)))
        (is (= :running (:task/status entity)))
        (is (= :active (:task/kanban-column entity)))
        (is (= "Implement X" (:task/description entity)))))))

(deftest task-transitions-emit-one-snapshot-per-change
  (let [stream (no-sink-stream)
        comp   (iface/create stream)
        tid    (random-uuid)
        wf-id  (random-uuid)]
    (iface/start! comp)
    (doseq [state [:pending :ready :running :completed]]
      (es/publish! stream (task-state-changed tid wf-id state)))
    (let [snaps (->> (supervisory-events stream)
                     (filter #(= :supervisory/task-node-upserted (:event/type %))))
          final (last snaps)]
      ;; One snapshot per distinct status transition (diff-based emit).
      (is (= 4 (count snaps)) "four transitions → four snapshots")
      (is (= :completed (:task/status (:supervisory/entity final))))
      (is (= :done (:task/kanban-column (:supervisory/entity final))))
      (is (some? (:task/completed-at (:supervisory/entity final))))
      (is (some? (:task/elapsed-ms (:supervisory/entity final)))))))

;------------------------------------------------------------------------------ DecisionCard (N5-δ3 §3.4)

(defn- cp-decision-created-event [decision-id agent-id summary]
  {:event/type :control-plane/decision-created
   :event/id (random-uuid)
   :event/timestamp (java.util.Date.)
   :event/version "1.0.0"
   :event/sequence-number 0
   :workflow/id (random-uuid)
   :cp/agent-id agent-id
   :cp/decision-id decision-id
   :cp/summary summary
   :message (str "Decision needed from " agent-id ": " summary)})

(defn- cp-decision-resolved-event [decision-id resolution]
  {:event/type :control-plane/decision-resolved
   :event/id (random-uuid)
   :event/timestamp (java.util.Date.)
   :event/version "1.0.0"
   :event/sequence-number 0
   :workflow/id (random-uuid)
   :cp/decision-id decision-id
   :cp/resolution resolution
   :message (str "Decision " decision-id " resolved: " resolution)})

(deftest decision-created-then-resolved-produces-two-snapshots
  (let [stream (no-sink-stream)
        comp   (iface/create stream)
        did    (random-uuid)
        aid    (random-uuid)]
    (iface/start! comp)
    (es/publish! stream (cp-decision-created-event  did aid "Approve the merge"))
    (es/publish! stream (cp-decision-resolved-event did "approve"))
    (let [snaps (->> (supervisory-events stream)
                     (filter #(= :supervisory/decision-upserted (:event/type %))))
          final (last snaps)]
      (is (= 2 (count snaps)) "pending + resolved = two snapshots")
      (is (= :resolved (:decision/status (:supervisory/entity final))))
      (is (= "approve"   (:decision/resolution (:supervisory/entity final))))
      (is (= aid         (:decision/agent-id (:supervisory/entity final)))
          "resolve event must preserve the agent-id from the create event"))))

;------------------------------------------------------------------------------ InterventionRequest

(defn- intervention-requested-event [workflow-id intervention-id]
  {:event/type :supervisory/intervention-requested
   :event/id (random-uuid)
   :event/timestamp (java.util.Date.)
   :event/version "1.0.0"
   :event/sequence-number 0
   :workflow/id workflow-id
   :intervention/id intervention-id
   :intervention/type :pause
   :intervention/target-type :workflow
   :intervention/target-id workflow-id
   :intervention/requested-by "operator@example.com"
   :intervention/request-source :tui
   :intervention/state :proposed
   :intervention/requested-at (java.util.Date.)
   :intervention/updated-at (java.util.Date.)
   :message "Pause requested"})

(defn- intervention-state-changed-event [workflow-id intervention-id next-state]
  {:event/type :supervisory/intervention-state-changed
   :event/id (random-uuid)
   :event/timestamp (java.util.Date.)
   :event/version "1.0.0"
   :event/sequence-number 0
   :workflow/id workflow-id
   :intervention/id intervention-id
   :intervention/state next-state
   :intervention/outcome {:paused true}
   :message (str "Intervention " intervention-id " → " (name next-state))})

(deftest intervention-events-produce-supervisory-intervention-snapshots
  (let [stream (no-sink-stream)
        comp   (iface/create stream)
        workflow-id (random-uuid)
        intervention-id (random-uuid)]
    (iface/start! comp)
    (es/publish! stream (intervention-requested-event workflow-id intervention-id))
    (es/publish! stream (intervention-state-changed-event workflow-id intervention-id :applied))
    (let [snaps (->> (supervisory-events stream)
                     (filter #(= :supervisory/intervention-upserted (:event/type %))))
          final (last snaps)]
      (is (= 2 (count snaps)) "request + applied state change should each emit a snapshot")
      (is (= :applied (get-in final [:supervisory/entity :intervention/state])))
      (is (= {:paused true} (get-in final [:supervisory/entity :intervention/outcome])))
      (is (= intervention-id (->> (iface/interventions comp)
                                  first
                                  :intervention/id))))))
