cthulhu/internal/tentacle/session.go

// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later

package tentacle

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"sync"
	"time"
)

type State string

const (
	Starting State = "starting"
	Ready    State = "ready"
	Active   State = "active"
	Stopping State = "stopping"
	Stopped  State = "stopped"
	Failed   State = "failed"
)

// Result is never delivered to an output backend by this package. The future
// reader output owner must revalidate context, lock, mode and interruption state
// when consuming it, including a result already buffered before cancellation.
type Result struct {
	Session string
	ID      string
	Value   json.RawMessage
	Err     error
}

type pendingCall struct {
	ctx       context.Context
	cancel    context.CancelFunc
	stopWatch func() bool
	result    chan Result
}

type outgoing struct {
	id    string // Empty for lifecycle and cancellation notifications.
	frame []byte
}

type Session struct {
	outputContext context.Context
	revokeOutput  context.CancelFunc

	mutex     sync.Mutex
	manager   *Manager
	spec      Spec
	identity  string
	ctx       context.Context
	cancel    context.CancelFunc
	state     State
	failure   error
	commands  []Command
	bindings  []BindingStatus
	sequence  uint64
	pending   map[string]*pendingCall
	outbound  chan outgoing
	ready     chan struct{}
	done      chan struct{}
	stop      chan struct{}
	revoked   chan struct{}
	callbacks sync.WaitGroup
}

func (session *Session) Ready() <-chan struct{} { return session.ready }

// Revoked closes as soon as dispatch is retired, before descendant cleanup.
func (session *Session) Revoked() <-chan struct{} { return session.revoked }
func (session *Session) Done() <-chan struct{}    { return session.done }
func (session *Session) Identity() string         { return session.identity }
func (session *Session) TentacleID() string       { return session.spec.ID }

// OutputContext expires synchronously on revocation, including after a reply
// has left the host's pending-call map. It never authorizes output by itself.
func (session *Session) OutputContext() context.Context { return session.outputContext }

func (session *Session) Status() (State, error) {
	session.mutex.Lock()
	defer session.mutex.Unlock()
	return session.state, session.failure
}

// Commands returns a copy of staged or active registrations. A caller may
// inspect Ready registrations but must not dispatch them before Activate.
func (session *Session) Commands() []Command {
	session.mutex.Lock()
	defer session.mutex.Unlock()
	return clone_commands(session.commands)
}

// Activate is the host publication barrier. The reader stages acknowledged
// native bindings before calling it; there are no external service exports.
func (session *Session) Activate() error {
	session.manager.mutex.Lock()
	defer session.manager.mutex.Unlock()
	occupied := session.manager.occupied_bindings()
	session.mutex.Lock()
	defer session.mutex.Unlock()
	if session.state != Ready || session.ctx.Err() != nil {
		return ErrUnavailable
	}
	if !session.notify_locked("initialized", struct {
		Session string `json:"session"`
	}{session.identity}) {
		return ErrBusy
	}
	session.bindings = resolve_bindings(session.spec.ID, session.commands, occupied)
	session.state = Active
	return nil
}

func (session *Session) Invoke(ctx context.Context, command string, arguments json.RawMessage) (<-chan Result, error) {
	if ctx == nil {
		return nil, errors.New("nil command context")
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if len(arguments) > MaxArgumentBytes {
		return nil, ErrLimit
	}
	if len(arguments) == 0 {
		arguments = json.RawMessage(`{}`)
	}
	if err := validate_json(arguments); err != nil {
		return nil, err
	}
	session.mutex.Lock()
	defer session.mutex.Unlock()
	if session.state != Active || session.ctx.Err() != nil {
		return nil, ErrUnavailable
	}
	known := false
	for _, registered := range session.commands {
		known = known || registered.ID == command
	}
	if !known {
		return nil, ErrUnavailable
	}
	return session.request_locked(ctx, "command.invoke", struct {
		Command   string          `json:"command"`
		Arguments json.RawMessage `json:"arguments"`
	}{command, arguments}, session.manager.limits.RequestTimeout)
}

func (session *Session) request_locked(parent context.Context, method string, params any, timeout time.Duration) (<-chan Result, error) {
	if len(session.pending) >= session.manager.limits.PendingPerProcess {
		return nil, ErrBusy
	}
	select {
	case session.manager.pending <- struct{}{}:
	default:
		return nil, ErrBusy
	}
	session.sequence++
	id := strconv.FormatUint(session.sequence, 10)
	frame, err := encode_request(id, method, params)
	if err != nil {
		<-session.manager.pending
		return nil, err
	}
	ctx, cancel := context.WithTimeout(parent, timeout)
	call := &pendingCall{ctx: ctx, cancel: cancel, result: make(chan Result, 1)}
	session.pending[id] = call
	session.callbacks.Add(1)
	call.stopWatch = context.AfterFunc(ctx, func() {
		defer session.callbacks.Done()
		session.cancel_call(id)
	})
	select {
	case session.outbound <- outgoing{id: id, frame: frame}:
		return call.result, nil
	default:
		session.finish_locked(id, nil, ErrBusy)
		return nil, ErrBusy
	}
}

func (session *Session) notify_locked(method string, params any) bool {
	frame, err := encode_request("", method, params)
	if err != nil {
		return false
	}
	select {
	case session.outbound <- outgoing{frame: frame}:
		return true
	default:
		return false
	}
}

func (session *Session) finish_locked(id string, value json.RawMessage, err error) {
	call := session.pending[id]
	if call == nil {
		return
	}
	delete(session.pending, id)
	if call.stopWatch() {
		session.callbacks.Done()
	}
	if contextError := call.ctx.Err(); contextError != nil && err == nil {
		err = contextError
		value = nil
	}
	call.cancel()
	<-session.manager.pending
	call.result <- Result{Session: session.identity, ID: id, Value: value, Err: err}
	close(call.result)
}

func (session *Session) cancel_call(id string) {
	session.mutex.Lock()
	defer session.mutex.Unlock()
	call := session.pending[id]
	if call == nil {
		return
	}
	err := call.ctx.Err()
	session.finish_locked(id, nil, err)
	// Without a cancellation acknowledgment contract, a timed-out child could
	// accumulate unlimited still-running work. Retire the whole session instead.
	// Explicit cancellation follows the same conservative rule in this slice.
	session.fail_locked(err)
}

func (session *Session) accept_reply(reply response) {
	session.mutex.Lock()
	defer session.mutex.Unlock()
	if session.state == Stopping || session.state == Stopped || session.state == Failed {
		return
	}
	if session.pending[reply.ID] == nil {
		// Unknown/duplicate responses are not allowed to keep a flood alive.
		session.fail_locked(ErrProtocol)
		return
	}
	var err error
	if reply.Error != nil {
		err = reply.Error
	}
	session.finish_locked(reply.ID, reply.Result, err)
}

func (session *Session) fail_locked(err error) {
	if session.state == Stopping || session.state == Stopped || session.state == Failed {
		return
	}
	session.state, session.failure = Failed, err
	close(session.revoked)
	session.revokeOutput()
	session.commands = nil
	session.bindings = nil
	for id := range session.pending {
		session.finish_locked(id, nil, err)
	}
	session.cancel()
}

func (session *Session) fail(err error) {
	session.mutex.Lock()
	defer session.mutex.Unlock()
	session.fail_locked(err)
}

// Disable revokes commands and retires replies synchronously; pipe shutdown and
// process cleanup happen on the supervisor. It is safe to call repeatedly.
func (session *Session) Disable() {
	session.mutex.Lock()
	defer session.mutex.Unlock()
	if session.state == Stopping || session.state == Stopped || session.state == Failed {
		return
	}
	session.state = Stopping
	close(session.revoked)
	session.revokeOutput()
	session.commands = nil
	session.bindings = nil
	for id := range session.pending {
		session.finish_locked(id, nil, ErrDisabled)
	}
	session.notify_locked("shutdown", nil)
	close(session.stop)
}