broker

package
v0.6.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 14, 2026 License: MIT Imports: 16 Imported by: 0

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

Constants

This section is empty.

Variables

View Source 
var ErrEvidenceLinkConflict = errors.New("evidence link conflicts with existing row")

ErrEvidenceLinkConflict reports a link key with a different stored hash.

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 ExtractFilePaths 

func ExtractFilePaths(toolUsesJSON string) []string

ExtractFilePaths parses the tool_uses JSON and returns all unique absolute file paths found. Recognizes both POSIX and Windows absolute path formats since agent payloads may use either regardless of host OS.

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 HasEvidenceLink(ctx context.Context, repoPath string, l EvidenceLink) (bool, error)

HasEvidenceLink reports whether the link key already exists.

func MissingEventIDs added in v0.6.0 

func MissingEventIDs(ctx context.Context, repoPath string, ids []string) (map[string]bool, error)

MissingEventIDs returns the given IDs that have no event row.

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 PruneStale instead.

func PruneStale added in v0.5.5 

func PruneStale(ctx context.Context, h *Handle, stale map[string]RepoStateReason) (int, error)

PruneStale removes registry entries the caller has already classified as confirmed-stale. The stale map is keyed by canonical path so it survives symlink normalisation; entries missing from the map are preserved. Callers must build the map from CheckRepoState so RepoStateUnknown verdicts are never pruned.

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 RepositoryIDForPath added in v0.6.0 

func RepositoryIDForPath(ctx context.Context, repoPath string) (string, error)

RepositoryIDForPath returns the repository UUID stored in its lineage database.

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.

func WriteEvidenceLinksToRepo added in v0.6.0 

func WriteEvidenceLinksToRepo(ctx context.Context, repoPath string, links []EvidenceLink) error

WriteEvidenceLinksToRepo atomically writes idempotent evidence links. Existing links must contain the same evidence hash.

Types

type ErrRepoStale added in v0.5.5 

type ErrRepoStale struct {
	RepoPath string
	Reason   RepoStateReason
}

ErrRepoStale means the target repo state is stale. Callers may drop these writes without reporting a hook failure.

func (*ErrRepoStale) Error added in v0.5.5 

func (e *ErrRepoStale) Error() string
type EvidenceLink struct {
	EventID      string
	EvidenceKind string
	EvidenceHash string
	GroupID      string
	CreatedAt    int64
}

EvidenceLink attaches a content-addressed evidence blob to an event row.

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 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 RepoStateReason added in v0.5.5 

type RepoStateReason string

RepoStateReason explains why a broker entry cannot receive events. Each value is a confirmed stale state that tidy may prune. 

const (
	// RepoStaleSemDirMissing: <repo>/.semantica does not exist.
	RepoStaleSemDirMissing RepoStateReason = "sem-dir-missing"
	// RepoStaleLineageDBMissing: <repo>/.semantica exists but lineage.db does not.
	RepoStaleLineageDBMissing RepoStateReason = "lineage-db-missing"
	// RepoStaleNoRepoRow: lineage.db opens but has no repositories row
	// for this path.
	RepoStaleNoRepoRow RepoStateReason = "no-repo-row"
	// RepoStaleSettingsDisabled: <repo>/.semantica/enabled marker is absent.
	RepoStaleSettingsDisabled RepoStateReason = "settings-disabled"
)

type RepoStateResult added in v0.5.5 

type RepoStateResult struct {
	Verdict RepoStateVerdict
	Reason  RepoStateReason
	Err     error
}

RepoStateResult is the result of CheckRepoState. Reason is set for stale entries; Err is set for unknown entries.

func CheckRepoState added in v0.5.5 

func CheckRepoState(ctx context.Context, repoPath string) RepoStateResult

CheckRepoState reports whether the broker can write to a repo. Unknown errors are not prunable.

The check is read-only: it may open lineage.db to look up the repositories row but never mutates it.

type RepoStateVerdict added in v0.5.5 

type RepoStateVerdict int

RepoStateVerdict summarizes whether a broker entry is usable. 

const (
	// RepoStateOK means the broker can safely route events to this repo.
	RepoStateOK RepoStateVerdict = iota
	// RepoStateStale means one of the four confirmed reasons applies;
	// pruning the registry entry is safe.
	RepoStateStale
	// RepoStateUnknown means an unexpected error (permission, IO, DB
	// corruption) blocked the check. Callers must not prune on Unknown.
	RepoStateUnknown
)

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.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.