dtoro
7a7d390718
Mechanical extraction of nomos internal components into plain-Go subpackages
per the hexagonal plan (ADR 0016 §3.1 rule 3):
turngate/ — per-session turn serialization (plan 2026-08-03 F1)
retrycap/ — per-turn run retry cap (maxRunRetries=3)
messagequeue/ — operator-message queue for busy-turn re-entry (F2)
assent/ — chat-assent detection (isAssent, isTypedConfirmation,
ExtractPendingApprovals), decoupled from agent via
[]string input instead of persistedCall
session/ — store (chat sessions, plan execution, DB persistence),
migration runner + local emitEvent to break adapter
dependency
internal/migrate/ — shared migration runner extracted from postgres pool,
used by both the oikos postgres adapter and session tests.
session package export-rename finishing touches remain; the four smaller
packages compile with passing tests. Depguard rules and ADR-0016 leaf-note
update deferred to a followup. VERSION 0.35.1.
83 lines
3.0 KiB
Go
83 lines
3.0 KiB
Go
package messagequeue
|
|
|
|
import (
|
|
"log/slog"
|
|
"sync"
|
|
)
|
|
|
|
// MaxQueuedPerSession caps a session's queue. A held turn plus unbounded
|
|
// enqueues would grow memory without limit; an operator nudging a long
|
|
// autonomous turn realistically queues only a handful, so a generous cap is
|
|
// pure insurance. Overflow drops the newest Enqueue and logs (the message is
|
|
// already persisted in the DB by handleChat before Enqueue, so it isn't lost
|
|
// from the transcript — it just won't auto-run).
|
|
const MaxQueuedPerSession = 20
|
|
|
|
// MessageQueue holds operator messages that arrived while a turn was already
|
|
// running for a session. Plan 2026-08-03 (F2): instead of rejecting the
|
|
// operator's message with "Nomos is still finishing a previous step… send it
|
|
// again", the message is queued and auto-run when the in-flight turn releases
|
|
// the session's turn-gate permit.
|
|
//
|
|
// The queue only schedules WHEN a turn runs, not WHETHER the message is stored
|
|
// — handleChat persists the user message before acquiring the gate, so a queued
|
|
// message is already in the transcript; this just makes sure a turn eventually
|
|
// acts on it.
|
|
//
|
|
// Draining is strictly one-at-a-time under the turn gate (see drainQueued in
|
|
// main.go), so this cannot stack concurrent turns — the exact hazard the gate
|
|
// itself exists to prevent. Background resumeSession callers never touch this
|
|
// queue; they keep their non-blocking skip.
|
|
type MessageQueue struct {
|
|
mu sync.Mutex
|
|
queue map[string][]string
|
|
}
|
|
|
|
func New() *MessageQueue {
|
|
return &MessageQueue{queue: map[string][]string{}}
|
|
}
|
|
|
|
// Enqueue appends a message to the back of the session's FIFO. Returns false
|
|
// (and logs) if the session is already at MaxQueuedPerSession — the caller's
|
|
// message is already persisted in the DB, so this only skips auto-running it.
|
|
func (q *MessageQueue) Enqueue(sessionID, msg string) bool {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if len(q.queue[sessionID]) >= MaxQueuedPerSession {
|
|
slog.Warn("nomos: message queue full; dropping auto-run for operator message", "session", sessionID, "cap", MaxQueuedPerSession)
|
|
return false
|
|
}
|
|
q.queue[sessionID] = append(q.queue[sessionID], msg)
|
|
return true
|
|
}
|
|
|
|
// Dequeue pops the next message from the front of the session's FIFO. Returns
|
|
// ok=false when empty.
|
|
func (q *MessageQueue) Dequeue(sessionID string) (string, bool) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
xs := q.queue[sessionID]
|
|
if len(xs) == 0 {
|
|
return "", false
|
|
}
|
|
m := xs[0]
|
|
q.queue[sessionID] = xs[1:]
|
|
return m, true
|
|
}
|
|
|
|
// RequeueFront pushes a message back to the front — used when a drainer popped
|
|
// a message but lost the race for the gate to a live turn; that turn's own
|
|
// release will drain it again.
|
|
func (q *MessageQueue) RequeueFront(sessionID, msg string) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
q.queue[sessionID] = append([]string{msg}, q.queue[sessionID]...)
|
|
}
|
|
|
|
// Peek reports the queued depth for a session (test/diagnostic helper).
|
|
func (q *MessageQueue) Peek(sessionID string) int {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
return len(q.queue[sessionID])
|
|
}
|