hypercore-kb / opencog-kb / language-learning / chatbot / continuous-daemon.scm

;
; continuous-daemon.scm - Background continuous learning daemon
;
; Uses unified COUNT + PMI approach for both UP (clustering) and DOWN (grounding)
; Processes queued messages, saves state, detects gaps continuously
;

(use-modules (ice-9 format))
(use-modules (ice-9 rdelim))
(use-modules (srfi srfi-1))

(primitive-load "continuous-learner.scm")

;; ═══════════════════════════════════════════════════════════════════
;; PERSISTENCE
;; ═══════════════════════════════════════════════════════════════════

(define *data-dir* "data")
(define *state-file* "data/continuous-state.scm")

(define (ensure-data-dir)
  (unless (file-exists? *data-dir*)
    (mkdir *data-dir*)))

(define (save-state!)
  "Save learner state to disk"
  (ensure-data-dir)
  (let ((port (open-output-file *state-file*)))
    (display ";; Continuous learner state\n" port)
    (display "(set! *unit-counts* '" port)
    (write *unit-counts* port)
    (display ")\n" port)
    (display "(set! *pair-counts* '" port)
    (write *pair-counts* port)
    (display ")\n" port)
    (display "(set! *total-observations* " port)
    (write *total-observations* port)
    (display ")\n" port)
    (close-port port)
    (display (format #f "[daemon] Saved: ~a units, ~a pairs\n"
                     (length *unit-counts*) (length *pair-counts*)))))

(define (load-state!)
  "Load learner state from disk"
  (when (file-exists? *state-file*)
    (display "[daemon] Loading state...\n")
    (primitive-load *state-file*)
    (display (format #f "[daemon] Loaded: ~a units, ~a pairs, ~a observations\n"
                     (length *unit-counts*) (length *pair-counts*) *total-observations*))))

;; ═══════════════════════════════════════════════════════════════════
;; QUEUE
;; ═══════════════════════════════════════════════════════════════════

(define *queue-file* "data/queue/pending.txt")
(define *queue-dir* "data/queue")

(define (ensure-queue-dir)
  (ensure-data-dir)
  (unless (file-exists? *queue-dir*)
    (mkdir *queue-dir*)))

(define (dequeue-batch! max-count)
  "Get up to max-count messages from queue"
  (if (not (file-exists? *queue-file*))
      '()
      (let ((messages '())
            (remaining '()))
        (let ((port (open-input-file *queue-file*)))
          (let loop ((count 0))
            (let ((line (read-line port)))
              (unless (eof-object? line)
                (if (< count max-count)
                    (set! messages (cons line messages))
                    (set! remaining (cons line remaining)))
                (loop (+ 1 count)))))
          (close-port port))
        ;; Write remaining back
        (if (null? remaining)
            (when (file-exists? *queue-file*)
              (delete-file *queue-file*))
            (let ((port (open-output-file *queue-file*)))
              (for-each (lambda (msg)
                          (display msg port)
                          (newline port))
                        (reverse remaining))
              (close-port port)))
        (reverse messages))))

(define (queue-length)
  (if (file-exists? *queue-file*)
      (let ((port (open-input-file *queue-file*)))
        (let loop ((count 0))
          (let ((line (read-line port)))
            (if (eof-object? line)
                (begin (close-port port) count)
                (loop (+ 1 count))))))
      0))

;; ═══════════════════════════════════════════════════════════════════
;; DAEMON LOOP
;; ═══════════════════════════════════════════════════════════════════

(define *batch-size* 50)
(define *batch-interval* 5)
(define *save-interval* 60)
(define *running* #t)

(define (process-batch!)
  "Process a batch of queued messages"
  (let ((messages (dequeue-batch! *batch-size*)))
    (when (not (null? messages))
      (display (format #f "[daemon] Processing ~a messages...\n" (length messages)))
      (for-each process-text! messages)
      (display (format #f "[daemon] Done. Units: ~a, Pairs: ~a, Vagueness: ~,2f\n"
                       (length *unit-counts*)
                       (length *pair-counts*)
                       (overall-vagueness)))
      ;; Save after each batch
      (save-state!))
    (length messages)))

(define (daemon-loop)
  "Main daemon loop"
  (display "[daemon] Starting continuous learning daemon...\n")
  (display (format #f "[daemon] Batch: ~a, Interval: ~as, Save: ~as\n"
                   *batch-size* *batch-interval* *save-interval*))

  ;; Load existing state
  (load-state!)

  (ensure-queue-dir)

  (let ((last-save (current-time)))
    (while *running*
      ;; Process pending messages
      (process-batch!)

      ;; Periodic save (even if no new messages)
      (when (> (- (current-time) last-save) *save-interval*)
        (save-state!)
        (set! last-save (current-time)))

      ;; Wait
      (sleep *batch-interval*)))

  (display "[daemon] Shutting down...\n")
  (save-state!))

;; ═══════════════════════════════════════════════════════════════════
;; ENTRY POINT
;; ═══════════════════════════════════════════════════════════════════

(when (and (defined? 'command-line)
           (> (length (command-line)) 1)
           (string=? (cadr (command-line)) "run"))
  (daemon-loop))