Amr Gharbeia
924bf8f479
Some checks failed
Deploy (Gitea) / deploy (push) Failing after 2s
- Restore (in-package :passepartout) to core-reason - Move *VAULT-MEMORY* back to core-skills - Fix ASDF and defstruct/defpackage ordering - Increase daemon timeout to 120s - Handshake: 0.5.0 Verified: daemon processes messages, TUI clean, gate trace works
225 lines
9.4 KiB
Common Lisp
225 lines
9.4 KiB
Common Lisp
(defpackage :passepartout.system-event-orchestrator
|
|
(:use :cl :passepartout)
|
|
(:export
|
|
:orchestrator-register-hook
|
|
:orchestrator-register-cron
|
|
:orchestrator-classify
|
|
:orchestrator-on-heartbeat
|
|
:orchestrator-bootstrap
|
|
:orchestrator-dispatch
|
|
:default-classifier
|
|
:parse-org-repeat
|
|
:*hook-registry*
|
|
:*cron-registry*
|
|
:*tier-classifier*))
|
|
|
|
(in-package :passepartout.system-event-orchestrator)
|
|
|
|
(defvar *hook-registry* (make-hash-table :test 'equal)
|
|
"Maps hook property string → list of gate function symbols.")
|
|
|
|
(defvar *cron-registry* (make-hash-table :test 'equal)
|
|
"Maps job name string → plist (:next-run :expression :repeat :action :tier).")
|
|
|
|
(defvar *tier-classifier* nil
|
|
"Optional function (context) → :reflex | :cognition | :reasoning.")
|
|
|
|
(defun default-classifier (context)
|
|
"Rule-based tier classification.
|
|
:reflex — file/shell operations, deterministic checks
|
|
:cognition — text processing, summarization, simple Q&A
|
|
:reasoning — planning, analysis, multi-step decisions"
|
|
(let* ((text (or (getf context :text) ""))
|
|
(lower (string-downcase text)))
|
|
(cond
|
|
((or (search "rm " lower)
|
|
(search "write-file" lower)
|
|
(search "shell" lower)
|
|
(search "verify-" lower))
|
|
:reflex)
|
|
((or (search "summarize" lower)
|
|
(search "list" lower)
|
|
(search "find " lower)
|
|
(search "what is" lower)
|
|
(search "search" lower))
|
|
:cognition)
|
|
(t :reasoning))))
|
|
|
|
(defun parse-org-repeat (timestamp-string)
|
|
(let* ((cleaned (string-trim '(#\< #\> #\Newline #\Tab) timestamp-string))
|
|
(parts (uiop:split-string cleaned :separator '(#\space)))
|
|
(repeat-part (ignore-errors (car (last parts)))))
|
|
(when (and repeat-part (uiop:string-prefix-p "+" repeat-part))
|
|
(let* ((rest (subseq repeat-part 1))
|
|
(num-end (position-if (lambda (c) (not (digit-char-p c))) rest))
|
|
(num (parse-integer (subseq rest 0 num-end)))
|
|
(unit-str (subseq rest num-end)))
|
|
(list (intern (string-upcase unit-str) :keyword) num)))))
|
|
|
|
(defun orchestrator-register-hook (hook-property gate-function)
|
|
"Registers a deterministic gate to fire when an Org node with
|
|
the #+HOOK: property matching HOOK-PROPERTY is modified."
|
|
(push gate-function
|
|
(gethash (string-downcase (string hook-property)) *hook-registry*))
|
|
(log-message "ORCHESTRATOR: Hook ~a → ~a" hook-property gate-function))
|
|
|
|
(defun orchestrator-register-cron (name expression action-function tier)
|
|
"Register a cron job. NAME is a keyword, EXPRESSION is an Org-mode
|
|
timestamp string with optional repeat. TIER is :reflex :cognition :reasoning."
|
|
(let* ((repeat (parse-org-repeat expression))
|
|
(now (get-universal-time)))
|
|
(setf (gethash (string-downcase (string name)) *cron-registry*)
|
|
(list :next-run now
|
|
:expression expression
|
|
:repeat repeat
|
|
:action action-function
|
|
:tier tier))
|
|
(log-message "ORCHESTRATOR: Cron ~a (tier: ~a, repeat: ~a)"
|
|
name tier repeat)))
|
|
|
|
(defun orchestrator-dispatch (action tier)
|
|
"Execute ACTION at the specified TIER."
|
|
(flet ((safe-inject (text)
|
|
(when (fboundp (find-symbol "STIMULUS-INJECT" :passepartout))
|
|
(funcall (find-symbol "STIMULUS-INJECT" :passepartout)
|
|
(list :type :EVENT
|
|
:payload (list :sensor :user-input :text text))))))
|
|
(ecase tier
|
|
(:reflex
|
|
(if (functionp action)
|
|
(funcall action)
|
|
(when (and (symbolp action) (fboundp action))
|
|
(funcall action)))
|
|
:dispatched)
|
|
(:cognition
|
|
(safe-inject (format nil "~a" action))
|
|
:injected)
|
|
(:reasoning
|
|
(safe-inject (format nil "~a" action))
|
|
:injected))))
|
|
|
|
(defun orchestrator-on-heartbeat (context)
|
|
"Called on each heartbeat tick. Checks and dispatches due cron jobs."
|
|
(declare (ignore context))
|
|
(let ((now (get-universal-time))
|
|
(due-jobs nil))
|
|
(maphash (lambda (name config)
|
|
(let ((next-run (getf config :next-run)))
|
|
(when (>= now next-run)
|
|
(push (cons name config) due-jobs))))
|
|
*cron-registry*)
|
|
(dolist (job due-jobs)
|
|
(let* ((name (car job))
|
|
(config (cdr job))
|
|
(action (getf config :action))
|
|
(tier (getf config :tier))
|
|
(repeat (getf config :repeat))
|
|
(result (orchestrator-dispatch action tier)))
|
|
(log-message "ORCHESTRATOR: Heartbeat dispatched ~a (tier: ~a) → ~a"
|
|
name tier result)
|
|
(when repeat
|
|
(let* ((unit (first repeat))
|
|
(value (second repeat))
|
|
(interval (case unit
|
|
(:d (* 86400 value))
|
|
(:w (* 604800 value))
|
|
(:m (* 2592000 value))
|
|
(t (* 3600 value)))))
|
|
(setf (getf (gethash name *cron-registry*) :next-run)
|
|
(+ now interval))))))
|
|
nil))
|
|
|
|
(defun orchestrator-scan-org-file (filepath)
|
|
"Scans a single Org file for HOOK and CRON properties in property drawers.
|
|
Returns a list of plists (:type :hook/:cron :name <str> :value <str>)."
|
|
(let ((results nil)
|
|
(in-properties nil)
|
|
(lines nil))
|
|
(handler-case
|
|
(setf lines (uiop:split-string (uiop:read-file-string filepath)
|
|
:separator '(#\Newline)))
|
|
(error (c)
|
|
(log-message "ORCHESTRATOR: Could not read ~a: ~a" filepath c)
|
|
(return-from orchestrator-scan-org-file nil)))
|
|
(dolist (line lines)
|
|
(let ((trimmed (string-trim '(#\Space) line)))
|
|
(when (string= trimmed ":PROPERTIES:")
|
|
(setf in-properties t))
|
|
(when (string= trimmed ":END:")
|
|
(setf in-properties nil))
|
|
(when in-properties
|
|
(cond
|
|
((uiop:string-prefix-p ":HOOK:" trimmed)
|
|
(let ((val (string-trim '(#\Space) (subseq trimmed 6))))
|
|
(push (list :type :hook :name val :file filepath) results)
|
|
(log-message "ORCHESTRATOR: Found hook ~a in ~a" val filepath)))
|
|
((uiop:string-prefix-p ":CRON:" trimmed)
|
|
(let ((val (string-trim '(#\Space) (subseq trimmed 6))))
|
|
(push (list :type :cron :name val :file filepath) results)
|
|
(log-message "ORCHESTRATOR: Found cron ~a in ~a" val filepath)))))))
|
|
(nreverse results)))
|
|
|
|
(defun orchestrator-bootstrap ()
|
|
"Scans all Org files in the memex for #+HOOK: and #+CRON: properties
|
|
and registers them. Scans ~/memex/projects/ and ~/memex/system/ by default."
|
|
(let* ((memex-dir (or (uiop:getenv "MEMEX_DIR")
|
|
(namestring (merge-pathnames "memex/" (user-homedir-pathname)))))
|
|
(scan-dirs (list (merge-pathnames "projects/" memex-dir)
|
|
(merge-pathnames "system/" memex-dir)))
|
|
(hook-count 0)
|
|
(cron-count 0))
|
|
(dolist (dir scan-dirs)
|
|
(handler-case
|
|
(let ((files (uiop:directory-files dir "*.org")))
|
|
(dolist (file files)
|
|
(let* ((path (namestring file))
|
|
(entries (orchestrator-scan-org-file path)))
|
|
(dolist (entry entries)
|
|
(let ((type (getf entry :type))
|
|
(name (getf entry :name)))
|
|
(cond
|
|
((eq type :hook)
|
|
(orchestrator-register-hook name
|
|
(lambda ()
|
|
(log-message "ORCHESTRATOR: Hook ~a fired" name))))
|
|
((eq type :cron)
|
|
(orchestrator-register-cron
|
|
(intern (string-upcase (format nil "cron-~a" name)) :keyword)
|
|
name
|
|
(lambda ()
|
|
(log-message "ORCHESTRATOR: Cron ~a fired" name))
|
|
:cognition))))
|
|
(if (eq (getf entry :type) :hook) (incf hook-count) (incf cron-count))))))
|
|
(error (c)
|
|
(log-message "ORCHESTRATOR: Could not scan ~a: ~a" dir c))))
|
|
(log-message "ORCHESTRATOR: Bootstrap complete (~d hooks, ~d cron jobs)"
|
|
hook-count cron-count)))
|
|
|
|
(defun events-start-heartbeat ()
|
|
"Starts the background heartbeat thread. v0.5.0: extracted from core-loop."
|
|
(let ((interval (or (ignore-errors (parse-integer (uiop:getenv "HEARTBEAT_INTERVAL"))) 60))
|
|
(auto-save (or (ignore-errors (parse-integer (uiop:getenv "MEMORY_AUTO_SAVE_INTERVAL"))) *memory-auto-save-interval*)))
|
|
(setf *memory-auto-save-interval* auto-save)
|
|
(setf *heartbeat-save-counter* 0)
|
|
(setf *heartbeat-thread*
|
|
(bt:make-thread
|
|
(lambda ()
|
|
(loop
|
|
(sleep interval)
|
|
(incf *heartbeat-save-counter*)
|
|
(when (>= *heartbeat-save-counter* (/ *memory-auto-save-interval* interval))
|
|
(setf *heartbeat-save-counter* 0)
|
|
(save-memory-to-disk))
|
|
(stimulus-inject
|
|
(list :type :EVENT :payload (list :sensor :heartbeat :unix-time (get-universal-time))))))
|
|
:name "passepartout-heartbeat"))))
|
|
|
|
(defskill :passepartout-system-event-orchestrator
|
|
:priority 80
|
|
:trigger (lambda (ctx)
|
|
(eq (getf (getf ctx :payload) :sensor) :heartbeat))
|
|
:deterministic (lambda (action context)
|
|
(declare (ignore action))
|
|
(orchestrator-on-heartbeat context)
|
|
nil))
|