broker

Documentation

Overview

Package broker implements the device-level event routing layer for Semantica.

The broker maintains a registry of enabled repositories in a JSON file (~/.semantica/repos.json) and routes agent events into the correct per-repo lineage databases based on touched file paths.

Index

• func CanonicalRepoPath(repoRoot string) string
• func Close(h *Handle) error
• func Deactivate(ctx context.Context, h *Handle, canonicalPath string) error
• func DefaultRegistryPath() (string, error)
• func EmitObservation(ctx context.Context, obs Observation)
• func EmitObservations(ctx context.Context, observations []Observation)
• func ExtractFilePaths(toolUsesJSON string) []string
• func GlobalBase() (string, error)
• func GlobalObjectsDir() (string, error)
• func PathBelongsToRepo(path, repoRoot string) bool
• func Prune(ctx context.Context, h *Handle) error
• func PruneConfirmedMissing(ctx context.Context, h *Handle) (int, error)
• func Register(ctx context.Context, h *Handle, repoPath, canonicalPath string) error
• func WriteEventsToRepo(ctx context.Context, repoPath string, events []RawEvent, ...) ([]string, error)
• type Handle
  • func Open(ctx context.Context, registryPath string) (*Handle, error)
• type Observation
• type RawEvent
• type RegisteredRepo
  • func ListActiveRepos(ctx context.Context, h *Handle) ([]RegisteredRepo, error)
  • func ListAllRepos(ctx context.Context, h *Handle) ([]RegisteredRepo, error)
• type RepoInfo
• type RepoMatch
  • func RouteEvents(events []RawEvent, repos []RegisteredRepo) []RepoMatch
  • func RouteNoPathEvents(events []RawEvent, repos []RegisteredRepo, sourceProjectPath string) *RepoMatch
• type StatusResult
  • func GetStatus(ctx context.Context) (*StatusResult, error)

Functions

func CanonicalRepoPath

func CanonicalRepoPath(repoRoot string) string

CanonicalRepoPath resolves symlinks and cleans the path for use as the canonical repo identity in the broker registry.

func Close

func Close(h *Handle) error

Close is a no-op retained for caller symmetry (defer broker.Close(bh)).

func Deactivate

func Deactivate(ctx context.Context, h *Handle, canonicalPath string) error

Deactivate marks a repository as inactive in the registry. Does not delete the entry - allows re-registration later.

func DefaultRegistryPath

func DefaultRegistryPath() (string, error)

DefaultRegistryPath returns the registry file path (<globalBase>/repos.json).

func EmitObservation

func EmitObservation(ctx context.Context, obs Observation)

EmitObservation writes a single observation. Convenience wrapper around EmitObservations for callers outside the broker write path.

func EmitObservations

func EmitObservations(ctx context.Context, observations []Observation)

EmitObservations writes a batch of observations to the global implementations DB in a single open/transaction/close cycle. Fail-open: returns on any error without propagating. No git calls. No rule evaluation.

func ExtractFilePaths

func ExtractFilePaths(toolUsesJSON string) []string

ExtractFilePaths parses the tool_uses JSON and returns all unique absolute file paths found. Only returns paths that look absolute (start with /).

func GlobalBase

func GlobalBase() (string, error)

GlobalBase returns the Semantica global directory. Defaults to ~/.semantica but can be overridden via the SEMANTICA_HOME env var (primarily for test isolation).

func GlobalObjectsDir

func GlobalObjectsDir() (string, error)

GlobalObjectsDir returns the path to the global blob store directory (<globalBase>/objects). Used by hook capture, worker reconciliation, and commit-msg catch-up as the single source blob store before events are routed and copied into per-repo stores.

func PathBelongsToRepo

func PathBelongsToRepo(path, repoRoot string) bool

PathBelongsToRepo returns true if path equals repoRoot or is a subdirectory of repoRoot, after canonicalization. This handles the common case where a tool session is launched from a subdirectory inside a registered repo.

func Prune

func Prune(ctx context.Context, h *Handle) error

Prune removes registry entries whose .semantica directory no longer exists. Called best-effort during status checks to clean up after manual deletions. Note: drops entries on any os.Stat error, not just ErrNotExist. For conservative cleanup, use PruneConfirmedMissing instead.

func PruneConfirmedMissing

func PruneConfirmedMissing(ctx context.Context, h *Handle) (int, error)

PruneConfirmedMissing removes registry entries only when their .semantica directory is confirmed missing (os.ErrNotExist). Permission errors and transient I/O failures keep the entry. Returns the number of entries removed.

func Register

func Register(ctx context.Context, h *Handle, repoPath, canonicalPath string) error

Register adds or reactivates a repository in the registry. canonicalPath should be the symlink-resolved, cleaned absolute path.

func WriteEventsToRepo

func WriteEventsToRepo(ctx context.Context, repoPath string, events []RawEvent, srcBlobStore *blobs.Store) ([]string, error)

WriteEventsToRepo writes broker-routed events into the target repo's lineage DB. Groups events by source and session, upserts the necessary records, and inserts events with INSERT OR IGNORE for idempotency.

If srcBlobStore is non-nil, payload blobs are copied from it into the target repo's blob store so that routed events retain their payload_hash and participate in line-level attribution.

Returns the session IDs that were created or updated in this repo.

Types

type Handle

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

Handle holds the loaded registry and the file path.

func Open

func Open(ctx context.Context, registryPath string) (*Handle, error)

Open loads (or initializes) the registry at registryPath.

type Observation

type Observation struct {
	Provider          string
	ProviderSessionID string
	ParentSessionID   string // empty if root
	SourceProjectPath string // raw SourceProjectPath from RawEvent
	TargetRepoPath    string
	EventTs           int64 // unix ms
}

Observation is a lightweight record of broker-routed activity for a session. Written to the global implementations DB for later reconciliation.

type RawEvent

type RawEvent struct {
	// Identity (content-addressed, stable across reruns).
	EventID   string
	SourceKey string
	Provider  string

	// Parsed fields from the provider's event format.
	Timestamp         int64
	Kind              string // user, assistant, tool_result, etc.
	Role              string // user, assistant, system, tool
	ToolUsesJSON      string // Serialized {"content_types":[...],"tools":[...]}
	Summary           string
	PayloadHash       string // CAS pointer to raw blob in blob store
	TokensIn          int64
	TokensOut         int64
	TokensCacheRead   int64
	TokensCacheCreate int64
	ProviderEventID   string

	// Normalized source position. Provider-specific: line number
	// (JSONL providers), message index (JSON providers). Used for
	// content-addressed event IDs and reconciliation bookmarks.
	SourcePosition int64

	// Routing data - absolute file paths extracted from tool_uses.
	// Used by RouteEvents to determine which repos this event belongs to.
	FilePaths []string

	// Turn and step provenance - links events to their prompt boundary
	// and specific tool invocation for dedup and drill-down.
	TurnID         string // turn that produced this event
	ToolUseID      string // stable provider tool call id
	ToolName       string // Write, Edit, Bash, Agent, etc.
	EventSource    string // "hook" or "transcript"
	ProvenanceHash string // CAS pointer to raw hook payload for backend reconstruction

	// Session context - needed to create/update sessions in target repos.
	ProviderSessionID string
	ParentSessionID   string // empty if no parent
	SessionStartedAt  int64
	SessionMetaJSON   string
	SourceProjectPath string // decoded project path for no-path fallback routing
	Model             string // LLM model name (e.g. "opus 4.6", "gemini-2.5-pro")
}

RawEvent is a parsed but unrouted event from a provider source. Contains all fields needed to write into any target repo's lineage DB.

type RegisteredRepo

type RegisteredRepo struct {
	RepoID        string `json:"repo_id"`
	Path          string `json:"path"`
	CanonicalPath string `json:"canonical_path"`
	EnabledAt     int64  `json:"enabled_at"`
	DisabledAt    *int64 `json:"disabled_at,omitempty"`
	Active        bool   `json:"active"`
}

RegisteredRepo represents a repository entry in the JSON registry.

func ListActiveRepos

func ListActiveRepos(ctx context.Context, h *Handle) ([]RegisteredRepo, error)

ListActiveRepos returns all currently active registered repositories.

func ListAllRepos

func ListAllRepos(ctx context.Context, h *Handle) ([]RegisteredRepo, error)

ListAllRepos returns all registered repositories (active and inactive).

type RepoInfo

type RepoInfo struct {
	Path          string `json:"path"`
	CanonicalPath string `json:"canonical_path"`
	EnabledAt     int64  `json:"enabled_at"`
	Active        bool   `json:"active"`
}

RepoInfo holds registration info for a single repository.

type RepoMatch

type RepoMatch struct {
	Repo   RegisteredRepo
	Events []RawEvent
}

RepoMatch pairs a registered repo with the events routed to it.

func RouteEvents

func RouteEvents(events []RawEvent, repos []RegisteredRepo) []RepoMatch

RouteEvents matches raw events to registered repos by file path containment.

Correctness rule: an event belongs to a repo if it touched one or more file paths under that repo's canonical root.

No-path fallback: events without file paths are not routed by this function. The caller must handle the fallback (e.g., route to the source repo when the provider session path matches a registered repo).

Deepest-match rule: each file path routes to the deepest (longest canonical path) matching repo. This prevents nested repos from leaking events to parent repos. An event that touches files in multiple repos (at different nesting levels) routes to each deepest match once.

Disabled repo rule: only active repos in the input slice are considered. The caller should pass ListActiveRepos output.

func RouteNoPathEvents

func RouteNoPathEvents(events []RawEvent, repos []RegisteredRepo, sourceProjectPath string) *RepoMatch

RouteNoPathEvents handles events that have no file paths by matching them to repos via the source's project path. This is a fallback heuristic for non-file events (e.g., pure text conversations), not a strong cross-repo routing rule.

sourceProjectPath should be the provider-specific project path that the session was launched from (e.g., the decoded Claude project directory).

type StatusResult

type StatusResult struct {
	Repos       []RepoInfo `json:"repos"`
	ActiveCount int        `json:"active_count"`
}

StatusResult holds the output of GetStatus.

func GetStatus

func GetStatus(ctx context.Context) (*StatusResult, error)

GetStatus collects broker-level diagnostics from the default registry. Returns empty status (not an error) if the registry does not exist yet.