knowledgesink

package
v0.0.166 Latest Latest
Warning

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

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	AutoSinkWindow   = 7 * 24 * time.Hour
	AutoSinkGrace    = 10 * time.Minute
	AutoSinkInterval = time.Hour
)

Auto-sink window and timing (daemon scheduler; CLI list uses the same window).

View Source 
const (
	SkipReasonInconclusive = "inconclusive" // no clear conclusion yet — advance checked only
	SkipReasonNoNew        = "no_new"       // nothing new vs hub — advance checked only
)

Skip reasons when has_new_knowledges is false.

View Source 
const (
	SourceAuto  = "auto"
	SourceUI    = "ui"
	SourceSlash = "slash"
)

Marcus sink trigger sources (CLI --source / daemon POST source).

View Source 
const (
	StateUnavailable = "unavailable"
	StateReady       = "ready"
	StateSunk        = "sunk"
	StateBehind      = "behind"
	StateRunning     = "running"
	StateFailed      = "failed"
)

Button / status states for callers (Marcus UI, CLI).

View Source 
const (
	PingInterval = 4 * time.Second
	StaleAfter   = time.Minute
)

Heartbeat while status=running. ~4s interval ⇒ 1m ≈ 14 missed pings.

View Source 
const LogsFileName = "logs.jsonl"

LogsFileName is the unified session debug transcript (append-only JSONL).

View Source 
const ResultSchemaExample = `` /* 278-byte string literal not displayed */

ResultSchemaExample is the JSON object shape for headless propose-only RunJSON.

View Source 
const ShipResultExample = `` /* 250-byte string literal not displayed */

ShipResultExample is the example object for the prompt Output section (new knowledges).

View Source 
const ShipResultSkipExample = `` /* 135-byte string literal not displayed */

ShipResultSkipExample is the skip contract (no hub writes).

View Source 
const TimeLayout = "2006-01-02 15:04:05 -0700 MST"

TimeLayout is human-readable local time with numeric offset + zone name.

Variables

This section is empty.

Functions

func AgentPrompt 

func AgentPrompt(in PromptInput) string

func AppendLog added in v0.0.162 

func AppendLog(sessionDir, stream, text, trigger string, sinkIndex int, now time.Time) error

AppendLog appends one JSONL record. Best-effort; errors are ignored by callers that also tee to a live UI log.

func AutoSinkWhy added in v0.0.162 

func AutoSinkWhy(v *StatusView) string

AutoSinkWhy is a short reason column for --show-auto-sinkable-sessions.

func BoolPtr added in v0.0.163 

func BoolPtr(v bool) *bool

BoolPtr returns a *bool for ShipResult literals in tests and helpers.

func CheckedCursor added in v0.0.165 

func CheckedCursor(m *Manifest) string

CheckedCursor is the sinkability watermark. Falls back to sunk for old manifests.

func EffectiveAgentRunner 

func EffectiveAgentRunner(mode Mode, runner string) (effective, mappedFrom string, err error)

EffectiveAgentRunner maps shared/prefs runner IDs to the launch family for mode.

Headless uses CLI runners (codex/grok); --open uses TTY (*-tty). Returns mappedFrom when an alias was applied (for notices).

func FilterMessagesAfter 

func FilterMessagesAfter(msgs []sessions.ChatMessage, after time.Time) []sessions.ChatMessage

func FormatMRTitle added in v0.0.162 

func FormatMRTitle(source, commitMsg string) string

FormatMRTitle builds merge_request.title: optional source prefix + agent line.

func FormatTime 

func FormatTime(t time.Time) string

func HasSinkHistory added in v0.0.162 

func HasSinkHistory(m *Manifest) bool

HasSinkHistory reports whether a manifest records a completed sink.

func HubGitUser 

func HubGitUser(opts Opts, hubDir string) (string, error)

HubGitUser returns stripped username from git config user.email in hub.

func IsAutoSinkable added in v0.0.162 

func IsAutoSinkable(v *StatusView) bool

IsAutoSinkable reports whether StatusView is eligible for an automatic sink. Enabled covers ready / behind / failed-with-work; sunk / running / unavailable are not.

func IsStaleRunning added in v0.0.162 

func IsStaleRunning(m *Manifest, now time.Time) bool

IsStaleRunning reports whether a running manifest's last_ping is older than StaleAfter. Empty last_ping while running is treated as stale only after callers have had a chance to write the first ping — if LastPing is empty, use Started-equivalent: treat as stale when we cannot prove a recent ping (conservative: stale).

func LogsPath added in v0.0.162 

func LogsPath(sessionDir string) string

func MRTitlePrefix added in v0.0.162 

func MRTitlePrefix(source string) string

MRTitlePrefix returns the Marcus trigger tag(s) for an MR title, or "" when source is empty/unknown (bare CLI).

func ManifestPath 

func ManifestPath(sessionDir string) string

func NewestMessageTime 

func NewestMessageTime(msgs []sessions.ChatMessage) time.Time

func ParseTime 

func ParseTime(s string) (time.Time, error)

func ReconcileStaleRunning added in v0.0.162 

func ReconcileStaleRunning(sessionDir string, m *Manifest, now time.Time) (bool, error)

ReconcileStaleRunning rewrites a stale running manifest to failed. Returns true if rewritten.

func Root 

func Root(stateDir string) string

func RunDir 

func RunDir(sessionDir string, index int) string

func SanitizeSessionID 

func SanitizeSessionID(id string) string

func SessionDir 

func SessionDir(stateDir, marcusSessionID string) string

func ShowPromptText 

func ShowPromptText(in PromptInput) string

ShowPromptText is the reviewable prompt (no disk writes).

func SingleLineMRTitle 

func SingleLineMRTitle(msg string) string

SingleLineMRTitle flattens a commit message for merge_request.title push option.

func StartPingLoop added in v0.0.162 

func StartPingLoop(ctx context.Context, sessionDir string, pid int, nowFn func() time.Time)

StartPingLoop refreshes last_ping every PingInterval until ctx is cancelled.

func SunkCursor added in v0.0.165 

func SunkCursor(m *Manifest) string

SunkCursor is the content watermark (prompt since). Empty if never shipped.

func TipAfterMax 

func TipAfterMax(tip, lastSinkMax time.Time) bool

TipAfterMax reports whether tip is strictly after lastMax (new messages). Unknown tip (zero) cannot claim "after cursor": if a cursor exists, treat as not-after so skip+checked-advance stays non-sinkable when Status cannot load tip. Zero tip and zero cursor returns true only as a degenerate case; BuildStatus uses neverSunk before TipAfterMax when both are empty.

func TouchPing added in v0.0.162 

func TouchPing(sessionDir string, pid int, now time.Time) error

TouchPing updates last_ping (and pid if > 0) while status is still running.

func UserFromEmail 

func UserFromEmail(email string) string

UserFromEmail strips the domain from an email (devuser@example.com → devuser).

func WithinAutoSinkWindow added in v0.0.162 

func WithinAutoSinkWindow(updatedAt, now time.Time, window time.Duration) bool

WithinAutoSinkWindow reports whether updatedAt is non-zero and within window of now.

func WriteManifest 

func WriteManifest(sessionDir string, m *Manifest) error

Types

type AgentFunc 

type AgentFunc func(ctx context.Context, opts agentrunapi.RunOpts, jsonSchema string) (jsonResult string, err error)

AgentFunc injects the propose agent (nil → agentrunapi.Run / RunJSON).

type AutoSinkableRow added in v0.0.162 

type AutoSinkableRow struct {
	SessionID string    `json:"session_id"`
	UpdatedAt time.Time `json:"updated_at"`
	State     string    `json:"state"`
	Why       string    `json:"why,omitempty"`
}

AutoSinkableRow is one session the hourly auto-sink would consider.

func FilterAndSortAutoSinkable added in v0.0.162 

func FilterAndSortAutoSinkable(
	sessions []SessionMeta,
	now time.Time,
	window time.Duration,
	statusFor func(id string) (*StatusView, error),
) ([]AutoSinkableRow, error)

FilterAndSortAutoSinkable returns sinkable sessions within the window, oldest UpdatedAt first. statusFor is called only for non-archived sessions inside the window; nil views or errors skip the row.

func PickOldestAutoSinkable added in v0.0.162 

func PickOldestAutoSinkable(
	sessions []SessionMeta,
	now time.Time,
	window time.Duration,
	statusFor func(id string) (*StatusView, error),
) (*AutoSinkableRow, error)

PickOldestAutoSinkable returns the first row from FilterAndSortAutoSinkable, or nil when none.

type GitRunner 

type GitRunner func(dir string, args ...string) (stdout, stderr string, err error)

GitRunner runs git -C dir args. Nil → exec.Command.

type LogLine added in v0.0.162 

type LogLine struct {
	TS        string `json:"ts"`
	Stream    string `json:"stream"` // stdout | stderr
	Text      string `json:"text"`
	SinkIndex *int   `json:"sink_index,omitempty"`
	Trigger   string `json:"trigger,omitempty"` // ui | auto | cli
}

LogLine is one JSON object per line in logs.jsonl.

type Manifest 

type Manifest struct {
	Version                        int      `json:"version"`
	MarcusSessionID                string   `json:"marcus_session_id"`
	GrokSessionID                  string   `json:"grok_session_id,omitempty"`
	LastSinkAt                     string   `json:"last_sink_at,omitempty"`
	LastSinkMaxMessageTimestamp    string   `json:"last_sink_max_message_timestamp,omitempty"`    // sunk / since
	LastCheckedMaxMessageTimestamp string   `json:"last_checked_max_message_timestamp,omitempty"` // sinkability
	NextSinkIndex                  int      `json:"next_sink_index"`
	LastSinkIndex                  int      `json:"last_sink_index"`
	Status                         string   `json:"status"`
	Error                          string   `json:"error,omitempty"`
	Pid                            int      `json:"pid,omitempty"`       // info only; not used for liveness
	LastPing                       string   `json:"last_ping,omitempty"` // TimeLayout; heartbeat while running
	LastPaths                      []string `json:"last_paths,omitempty"`
	LastHubPaths                   []string `json:"last_hub_paths,omitempty"`
	LastBranch                     string   `json:"last_branch,omitempty"`
	LastMRURL                      string   `json:"last_mr_url,omitempty"`
	LastCommit                     string   `json:"last_commit,omitempty"`
}

Manifest is session-level cursor/status under knowledge-sink/<id>/manifest.json. status (+ last_ping) is the single source of truth for UI and all callers.

Two tip watermarks:

  • LastSinkMaxMessageTimestamp (sunk): last tip through which content was shipped; injected as prompt since on the next Run.
  • LastCheckedMaxMessageTimestamp (checked): last tip Status/auto-pick already considered; skips advance this without moving sunk.

Empty checked migrates to sunk for older manifests.

func LatchRunning added in v0.0.162 

func LatchRunning(sessionDir, marcusSessionID string, pid int, now time.Time) (*Manifest, error)

LatchRunning sets status=running, refreshes last_ping, optional pid (0 = leave / clear).

func LoadManifest 

func LoadManifest(sessionDir string) (*Manifest, error)

type MessagesFunc 

type MessagesFunc func(grokHome, sessionID string, opts *sessions.MessagesOpts) (*sessions.MessagesResult, error)

MessagesFunc injects sessions.Messages for tests (nil → library). Grok tip only.

type Mode 

type Mode string

Mode selects how the propose agent is launched. 

const (
	ModeHeadless Mode = "headless" // default: CLI codex/grok via agentui
	ModeOpen     Mode = "open"     // interactive open (TTY)
)

type Opts 

type Opts struct {
	StateDir      string // required: root that contains knowledge-sink/
	HubDir        string // knowledge-base-hub checkout (required to run agent); caller sets cwd
	SessionID     string // Marcus / agent-run session id
	GrokSessionID string // optional direct runner session id (skips resolve when set)
	Mode          Mode   // empty → headless
	DryRun        bool
	ShowPrompt    bool
	// CreateMR: agent applies hub writes + result.json; host commits/pushes/creates MR.
	CreateMR bool
	// AutoMergeMR implies CreateMR; after MR push, ff-merge and push origin/master.
	AutoMergeMR bool
	// AllowRunning skips the "already sinking" guard. Daemon uses this after it
	// latches manifest status=running for UI before the worker goroutine starts.
	AllowRunning bool
	// Source tags the Marcus trigger for MR title prefixes (auto|ui|slash).
	// Empty → no prefix (bare CLI).
	Source string
	// SkipSessionDirProbe skips Codex/Grok session-file walks when computing
	// Status. For list/auto-pick, a resolved runner session id counts as
	// having content (Codex Find was ~0.3–1s per session via full tree scan).
	SkipSessionDirProbe bool

	// AgentRunner / Model / ModelReasoningEffort are passed to agentrunapi.RunOpts.
	// Empty AgentRunner → library default (grok-tty); callers that want Marcus/
	// shared prefs should resolve before Run.
	AgentRunner          string
	Model                string
	ModelReasoningEffort string
	// Verbose prints concrete agent-run / Codex argv and post-agent ship
	// progress (git commit/push/MR) on Stderr.
	Verbose bool
	// Stderr receives verbose notices (nil → os.Stderr when Verbose).
	Stderr io.Writer

	Driver    agentdriver.Driver
	StoreHome string // agent-run job store; empty → StateDir/knowledge-sink-jobs/<id>

	MessagesFn          MessagesFunc
	AgentFn             AgentFunc
	ResolveFn           ResolveRunnerFunc
	ResolveSessionDirFn ResolveSessionDirFunc
	GitFn               GitRunner
	NowFn               func() time.Time
}

Opts drives Status and Run.

type PriorSinkContext 

type PriorSinkContext struct {
	LastSinkAt   string
	LastCursor   string
	PriorRunDirs []string // absolute sink-N dirs
	LastHubPaths []string
	LastPaths    []string // relative under session sink dir
	HasPrior     bool
}

PriorSinkContext is previous sink bookkeeping for dedup / incremental prompts.

type PromptInput 

type PromptInput struct {
	MarcusSessionID  string
	Runner           string // grok-tty | codex-tty | …
	RunnerSessionID  string
	RunnerSessionDir string // absolute primary source
	Since            string // sunk cursor (last_sink_max_message_timestamp), optional
	SinkIndex        int
	ProposalPath     string // absolute path for proposal.md
	Prior            PriorSinkContext

	// CreateMR mode: agent applies hub writes and emits result.json for host git.
	CreateMR       bool
	ResultJSONPath string // absolute sink-N/result.json
	GitUser        string // email local-part for branch prefix
	BranchDate     string // YYYY-MM-DD
}

PromptInput fills AgentPrompt / ShowPromptText.

type ResolveGrokFunc 

type ResolveGrokFunc = ResolveRunnerFunc

ResolveGrokFunc is kept as an alias name for older call sites.

type ResolveRunnerFunc 

type ResolveRunnerFunc func(marcusSessionID string) (runner, runnerSessionID string, err error)

ResolveRunnerFunc resolves Marcus/agent-run session → (runner, runnerSessionID).

type ResolveSessionDirFunc 

type ResolveSessionDirFunc func(runner, runnerSessionID string) (dir string, err error)

ResolveSessionDirFunc resolves runner session → absolute session dir (nil → grok/codex libs).

type RunResult 

type RunResult struct {
	OK               bool     `json:"ok"`
	DryRun           bool     `json:"dry_run,omitempty"`
	ShowPrompt       bool     `json:"show_prompt,omitempty"`
	CreateMR         bool     `json:"create_mr,omitempty"`
	AutoMergeMR      bool     `json:"auto_merge_mr,omitempty"`
	Mode             string   `json:"mode,omitempty"`
	SessionID        string   `json:"session_id,omitempty"`
	GrokSessionID    string   `json:"grok_session_id,omitempty"`
	Runner           string   `json:"runner,omitempty"`
	RunnerSessionDir string   `json:"runner_session_dir,omitempty"`
	SinkIndex        int      `json:"sink_index,omitempty"`
	DeltaCount       int      `json:"delta_count,omitempty"`
	ProposalPath     string   `json:"proposal_path,omitempty"`
	ResultJSONPath   string   `json:"result_json_path,omitempty"`
	SessionDir       string   `json:"session_dir,omitempty"` // knowledge-sink/<id>
	HubDir           string   `json:"hub_dir,omitempty"`
	Prompt           string   `json:"prompt,omitempty"`
	HubPaths         []string `json:"hub_paths,omitempty"` // proposed / committed hub paths
	Branch           string   `json:"branch,omitempty"`
	Commit           string   `json:"commit,omitempty"`
	MRURL            string   `json:"mr_url,omitempty"`
	Merged           bool     `json:"merged,omitempty"`
	MergedAt         string   `json:"merged_at,omitempty"`
	Warning          string   `json:"warning,omitempty"`
	LastSinkAt       string   `json:"last_sink_at,omitempty"`
	CursorTimestamp  string   `json:"cursor_timestamp,omitempty"`
	CursorAdvanced   bool     `json:"cursor_advanced,omitempty"`
	HasNewKnowledges *bool    `json:"has_new_knowledges,omitempty"`
	SkipReason       string   `json:"skip_reason,omitempty"`
	Error            string   `json:"error,omitempty"`
}

RunResult is Run output (including dry-run / show-prompt).

func Run 

func Run(ctx context.Context, opts Opts) (*RunResult, error)

Run performs a sink (propose-only by default, or create-mr / dry-run / show-prompt). Propose-only does not advance tip cursors. Create-MR advances checked on finish; sunk advances only when knowledges are shipped.

type SessionMeta added in v0.0.162 

type SessionMeta struct {
	ID        string
	UpdatedAt time.Time
	Archived  bool
}

SessionMeta is one Marcus/local-bot session for auto-sink selection.

type ShipCommitFiles 

type ShipCommitFiles struct {
	Add    []string `json:"add,omitempty"`
	Update []string `json:"update,omitempty"`
	Delete []string `json:"delete,omitempty"`
}

ShipCommitFiles is the agent-written hub path set for --create-mr shipping. add/update paths must exist on disk; delete paths must be absent and still tracked in git. Empty buckets may be omitted; at least one path overall when has_new_knowledges is true.

func (ShipCommitFiles) AllPaths 

func (f ShipCommitFiles) AllPaths() []string

AllPaths returns add ∪ update ∪ delete in that order (no dedupe across buckets).

func (*ShipCommitFiles) UnmarshalJSON 

func (f *ShipCommitFiles) UnmarshalJSON(data []byte) error

UnmarshalJSON rejects legacy flat string arrays with a clear contract error.

type ShipGitResult 

type ShipGitResult struct {
	Branch     string
	Commit     string
	MRURL      string
	Merged     bool
	MergedAt   string // short tip of origin/<target> after auto-merge
	PushOption bool
	Warning    string
}

ShipGitResult is the outcome of host-side commit/push/(auto-merge).

func ShipToMR 

func ShipToMR(opts Opts, hubDir string, ship *ShipResult, autoMerge bool) (*ShipGitResult, error)

ShipToMR stashes agent edits, branches from origin/master, commits, pushes with merge_request push options, optionally ff-merges into origin/master.

type ShipResult 

type ShipResult struct {
	// HasNewKnowledges is required. Pointer so missing JSON is distinct from false.
	HasNewKnowledges *bool           `json:"has_new_knowledges"`
	SkipReason       string          `json:"skip_reason,omitempty"` // inconclusive|no_new when false
	GitCommitMsg     string          `json:"git_commit_msg"`
	GitBranchName    string          `json:"git_branch_name"`
	GitCommitFiles   ShipCommitFiles `json:"git_commit_files"`
}

ShipResult is the agent-written contract for --create-mr shipping.

func ReadValidateShipResult 

func ReadValidateShipResult(path, hubDir string) (*ShipResult, error)

ReadValidateShipResult loads and validates sink-N/result.json against hubDir.

func (*ShipResult) HasNew added in v0.0.163 

func (sr *ShipResult) HasNew() bool

HasNew reports whether the agent marked new knowledges (false if nil/missing).

type SinkedRow added in v0.0.162 

type SinkedRow struct {
	SessionID  string    `json:"session_id"`
	LastSinkAt time.Time `json:"last_sink_at"`
	Status     string    `json:"status"` // idle|running|failed (manifest)
	LastMRURL  string    `json:"last_mr_url,omitempty"`
}

SinkedRow is one session that has completed a sink (manifest last_sink_at set).

func ListSinkedSessions added in v0.0.162 

func ListSinkedSessions(stateDir string) ([]SinkedRow, error)

ListSinkedSessions scans stateDir/knowledge-sink/*/manifest.json and returns sessions with last_sink_at set, newest last_sink_at first.

type StatusResult 

type StatusResult struct {
	OK            bool        `json:"ok"`
	SessionID     string      `json:"session_id,omitempty"`
	GrokSessionID string      `json:"grok_session_id,omitempty"` // runner session id (legacy json name)
	Sink          *StatusView `json:"sink,omitempty"`
	SessionDir    string      `json:"session_dir,omitempty"` // knowledge-sink/<id>
	LastSinkAt    string      `json:"last_sink_at,omitempty"`
	LastPaths     []string    `json:"last_paths,omitempty"`
	LastMRURL     string      `json:"last_mr_url,omitempty"`
	Error         string      `json:"error,omitempty"`
}

StatusResult is Status output.

func Status 

func Status(ctx context.Context, opts Opts) (*StatusResult, error)

Status computes sink button state for a session.

type StatusView 

type StatusView struct {
	State    string `json:"state"` // unavailable|ready|sunk|behind|running|failed
	Label    string `json:"label"`
	Enabled  bool   `json:"enabled"`
	Help     string `json:"help,omitempty"`
	Error    string `json:"error,omitempty"`
	NewCount int    `json:"new_count,omitempty"`
}

StatusView is the sink button / status snapshot.

func BuildStatus 

func BuildStatus(manifest *Manifest, tip time.Time, total int, grokOK bool, grokHelp string) *StatusView

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.