peermsg

Documentation

Overview

docs.go writes the peer-messaging capability document (RMOTE.md) and ensures the global agent instruction files reference it. Called during daemon startup so every agent session on the host auto-discovers peer messaging without per-project or per-agent configuration.

Strategy:

• ~/.rmote/RMOTE.md is the single source of truth (rewritten on every daemon start so content stays current).
• Claude Code: @import line in ~/.claude/CLAUDE.md (one line, Claude Code follows @path imports natively).
• All other agents: marker-delimited section injected into their global instructions file. The markers let the daemon update the section without clobbering user content.

Package peermsg implements cross-agent inter-session messaging for rmoted. Agents discover peers via the `rmoted peers` CLI and send messages via `rmoted message`. The daemon brokers delivery: direct PTY injection when an agent is foreground, or filesystem mailbox queueing when the target is at a shell prompt or offline.

Security posture: message bodies are untrusted DATA. All injected text is single-line (multiline triggers paste-confirm on some agents) with C0/ESC bytes stripped. The audit log records from/to/size/path only — never body content (which could carry exfiltrated secrets).

Index

• Constants
• Variables
• func AgentAbbrev(agent string) string
• func DeleteStaging(path string)
• func DeriveNames(inputs []NameInput) map[string]string
• func EnsureAgentDocs() error
• func FilePrompt(path, fromName string) string
• func FormatConfirm(targetName string) string
• func FormatOneLine(msg PeerMessage) string
• func HasMailbox(sessionID string) bool
• func StagingExists(path string) bool
• func SweepStaging(maxAge time.Duration) []string
• func WriteMailbox(sessionID string, msg PeerMessage) error
• func WriteStaging(text string) (string, error)
• func WriteTaskStaging(taskID, text string) (string, error)
• type Interceptor
  • func NewInterceptor(route RouteFunc) *Interceptor
  • func (in *Interceptor) IsPeerMode(sessionID string) bool
  • func (in *Interceptor) ProcessInput(sessionID string, data []byte) (passThrough, echo []byte)
  • func (in *Interceptor) Reset(sessionID string)
• type NameInput
• type PeerMessage
  • func DrainMailbox(sessionID string) ([]PeerMessage, error)
• type RateLimiter
  • func NewRateLimiter() *RateLimiter
  • func (rl *RateLimiter) Allow(from, to string) bool
• type RouteFunc

Constants

const MaxBodySize = 32 * 1024 // 32 KB

MaxBodySize is the hard cap on message text length.

Variables

var NamePattern = regexp.MustCompile(`#[a-z]{2,3}\d+`)

NamePattern matches a short-name token like #cc1, #cx12, #gk3.

Functions

func AgentAbbrev

func AgentAbbrev(agent string) string

AgentAbbrev returns a 2-letter abbreviation for an agent kind, used to build short names like #cc1, #cx1. Covers all catalog agents; unknown agents get the first two alphanumeric chars of their ID.

func DeleteStaging

func DeleteStaging(path string)

DeleteStaging removes a staging file after confirmed delivery. Best-effort.

func DeriveNames

func DeriveNames(inputs []NameInput) map[string]string

DeriveNames returns a sessionID → #name map for the given sessions. Names are deterministic: sort by session ID → group by agent abbreviation → number within group (#cc1, #cc2, #cx1, ...).

func EnsureAgentDocs

func EnsureAgentDocs() error

EnsureAgentDocs writes ~/.rmote/RMOTE.md and patches every agent's global instruction file. Idempotent — safe on every daemon startup. Errors are collected per-target; a failure on one agent does not block the others.

func FilePrompt

func FilePrompt(path, fromName string) string

FilePrompt returns a short "read this file" instruction for the large-message delivery path. The message text is written to a staging file; only this prompt is injected into the PTY.

func FormatConfirm

func FormatConfirm(targetName string) string

FormatConfirm returns the confirmation string for a successfully routed message. Used by RouteFunc implementations.

func FormatOneLine

func FormatOneLine(msg PeerMessage) string

FormatOneLine produces a single-line PTY-injectable string. Multiline triggers paste-confirm on some agents regardless of byte size, so the entire message stays on one line.

Format: [#cc2] message text — the sender's short name in brackets followed by the message. The recipient sees who sent it and can reply to that #name. Reply instructions live in RMOTE.md (auto-deployed by EnsureAgentDocs), not in the message text — keeps messages clean for human readers.

func HasMailbox

func HasMailbox(sessionID string) bool

HasMailbox returns true if the session has queued messages.

func StagingExists

func StagingExists(path string) bool

StagingExists verifies that a mailbox-supplied path remains an existing regular file inside rmote's private staging directory.

func SweepStaging

func SweepStaging(maxAge time.Duration) []string

SweepStaging removes staging files older than maxAge. Called periodically by the server's sweeper goroutine (H2: staging files must not leak forever).

func WriteMailbox

func WriteMailbox(sessionID string, msg PeerMessage) error

WriteMailbox queues a message for a session that is currently at a shell prompt (no agent foreground) or offline. The message is written as JSON to ~/.rmote/mailbox/{sessionID}/{timestamp}.msg (mode 0600).

func WriteStaging

func WriteStaging(text string) (string, error)

WriteStaging writes message text to a temporary file for the file-based delivery path (used when the message exceeds the direct-inject threshold). Returns the absolute path the agent should read.

Uses nanosecond timestamps + random suffix to avoid collisions, 0600 perms, and a separate directory from handoffs. Caller MUST DeleteStaging after confirmed delivery.

func WriteTaskStaging

func WriteTaskStaging(taskID, text string) (string, error)

Types

type Interceptor

type Interceptor struct {
	// contains filtered or unexported fields
}

Interceptor manages per-session peer-messaging input interception.

func NewInterceptor

func NewInterceptor(route RouteFunc) *Interceptor

NewInterceptor creates an interceptor with the given route callback.

func (*Interceptor) IsPeerMode

func (in *Interceptor) IsPeerMode(sessionID string) bool

IsPeerMode returns true if the session is currently in peer-messaging mode.

func (*Interceptor) ProcessInput

func (in *Interceptor) ProcessInput(sessionID string, data []byte) (passThrough, echo []byte)

ProcessInput processes a chunk of client input before it reaches the PTY. Returns passThrough (bytes to write to PTY) and echo (bytes to show the client via InjectOutput). When passThrough is nil, all input was intercepted.

Peer mode triggers when a chunk's first byte is '#' (0x23). This covers both single-keystroke input (# typed alone) and batch submission (#cc1 hello\r sent as one chunk). Subsequent chunks are buffered until \r (submit), Escape/Ctrl+C (cancel), or Backspace (edit).

func (*Interceptor) Reset

func (in *Interceptor) Reset(sessionID string)

Reset clears the interceptor state for a session (e.g. on session destroy).

type NameInput

type NameInput struct {
	SessionID string
	Agent     string // agent kind string ("claude", "codex", "shell", ...)
}

NameInput is the minimal info needed to derive a short name for a session.

type PeerMessage

type PeerMessage struct {
	TaskID      string `json:"task_id,omitempty"`
	StagingPath string `json:"staging_path,omitempty"`
	From        string `json:"from"`                 // RMOTE_SESSION_ID of sender
	FromAgent   string `json:"from_agent,omitempty"` // derived server-side from fgagent
	FromName    string `json:"from_name,omitempty"`  // sender's short name (#cc1)
	FromTitle   string `json:"from_title,omitempty"` // sanitized session title
	Text        string `json:"text"`                 // message body
	Timestamp   int64  `json:"timestamp"`            // unix nanos
}

PeerMessage is one inter-session message.

func DrainMailbox

func DrainMailbox(sessionID string) ([]PeerMessage, error)

DrainMailbox reads and removes all queued messages for a session, sorted oldest-first. The caller MUST hold a per-session mutex to prevent racing with concurrent live delivery (red-team M7).

type RateLimiter

type RateLimiter struct {
	// contains filtered or unexported fields
}

RateLimiter enforces per-pair and global rate limits.

func NewRateLimiter

func NewRateLimiter() *RateLimiter

NewRateLimiter creates a rate limiter ready for concurrent use.

func (*RateLimiter) Allow

func (rl *RateLimiter) Allow(from, to string) bool

Allow returns true if a message from→to is within both per-pair and global limits. Thread-safe.

type RouteFunc

type RouteFunc func(fromSessionID, targetName, message string) string

RouteFunc delivers a peer message. Returns a confirmation string shown to the sender (e.g. "[sent to #cx1]" or "[error: ...]").