ccbroker

May 15, 2026

Changes in this version

+ type AgentTurn struct

+ EnqueuedAt time.Time

+ ErrorMsg sql.NullString

+ FinishedAt sql.NullTime

+ ID string

+ IMChannelID sql.NullString

+ IMUserID sql.NullString

+ Metadata json.RawMessage

+ SessionID string

+ StartedAt sql.NullTime

+ State string

+ UserEventID string

+ UserMessage string

+ WorkspaceID string

type ProcessTurnRequest

+ TurnID string

type Server

+ func (s *Server) Shutdown(ctx context.Context) error

+ func (s *Server) Start(ctx context.Context) error

type Store

+ func (s *Store) CountPending(ctx context.Context, sessionID string) (int, error)

+ func (s *Store) EnqueueTurn(ctx context.Context, t AgentTurn) error

+ func (s *Store) GetTurn(ctx context.Context, turnID string) (*AgentTurn, error)

+ func (s *Store) GetTurnEvents(ctx context.Context, turnID string, sinceSeqNum int64) ([]TurnEvent, error)

+ func (s *Store) InsertEventsWithTurn(ctx context.Context, sessionID string, epoch int, turnID string, ...) ([]InsertedEvent, error)

+ func (s *Store) ListSessionTurns(ctx context.Context, sessionID string, limit int) ([]AgentTurn, error)

+ func (s *Store) ListSessionsWithPending(ctx context.Context) ([]string, error)

+ func (s *Store) MarkTurnCancelled(ctx context.Context, turnID string) error

+ func (s *Store) MarkTurnDone(ctx context.Context, turnID string) error

+ func (s *Store) MarkTurnFailed(ctx context.Context, turnID, errMsg string) error

+ func (s *Store) MarkTurnRunning(ctx context.Context, turnID string) error

+ func (s *Store) PickNextPending(ctx context.Context, sessionID string) (*AgentTurn, error)

+ func (s *Store) ResetRunningToQueued(ctx context.Context) (int, error)

type StreamClientEvent

+ TurnID string

+ type TurnEvent struct

+ CreatedAt time.Time

+ EventID string

+ EventType string

+ Payload json.RawMessage

+ SeqNum int64

May 3, 2026

Changes in this version

type Config

+ AgentserverInternalURL string

+ InternalAPISecret string

+ LLMProxyURL string

type ProcessTurnRequest

+ Metadata TurnMetadata

+ type TurnMetadata struct

+ ChannelType string

+ CreatorUserID string

+ Model string

+ PermissionMode string

+ PreferredExecutorID string

+ TurnKind string

May 3, 2026

Changes in this version

type Config

+ S3AccessKeyID string

+ S3Bucket string

+ S3Endpoint string

+ S3PathStyle bool

+ S3Region string

+ S3SecretAccessKey string

+ S3UseSSL bool

Apr 17, 2026

Changes in this version

+ var ErrJWTExpired = errors.New("jwt expired")

+ var ErrJWTMalformed = errors.New("jwt malformed")

+ var ErrJWTSignature = errors.New("jwt signature invalid")

+ func EpochFromContext(ctx context.Context) int

+ func IssueWorkerJWT(secret []byte, claims WorkerJWTClaims) (string, error)

+ func SessionIDFromContext(ctx context.Context) string

+ func WorkspaceIDFromContext(ctx context.Context) string

+ type BoundedUUIDSet struct

+ func NewBoundedUUIDSet(capacity int) *BoundedUUIDSet

+ func (s *BoundedUUIDSet) Add(uuid string) bool

+ type BridgeResponse struct

+ APIBaseURL string

+ ExpiresIn int

+ WorkerEpoch int

+ WorkerJWT string

+ type CCWorker struct

+ ID string

+ Process *exec.Cmd

+ SessionID string

+ StartedAt time.Time

+ Status string

+ WorkspaceID string

+ type Config struct

+ AgentserverURL string

+ DatabaseURL string

+ ExecutorRegistryURL string

+ IMBridgeSecret string

+ IMBridgeURL string

+ JWTSecret []byte

+ LogLevel slog.Level

+ OpenVikingAPIKey string

+ OpenVikingURL string

+ Port string

+ func LoadConfigFromEnv() (Config, error)

+ type DedupRegistry struct

+ func NewDedupRegistry() *DedupRegistry

+ func (r *DedupRegistry) GetOrCreate(sessionID string) *BoundedUUIDSet

+ type EventBatchItem struct

+ Ephemeral bool

+ Payload json.RawMessage

+ type EventBatchRequest struct

+ Events []EventBatchItem

+ WorkerEpoch int

+ type EventInput struct

+ Ephemeral bool

+ EventID string

+ Payload json.RawMessage

+ type HeartbeatRequest struct

+ WorkerEpoch int

+ type InsertedEvent struct

+ EventID string

+ SeqNum int64

+ type InternalEventBatchItem struct

+ AgentID string

+ IsCompaction bool

+ Payload json.RawMessage

+ type InternalEventBatchRequest struct

+ Events []InternalEventBatchItem

+ WorkerEpoch int

+ type InternalEventInput struct

+ AgentID string

+ EventType string

+ IsCompaction bool

+ Payload json.RawMessage

+ type JSONRPCRequest struct

+ ID *json.RawMessage

+ JSONRPC string

+ Method string

+ Params json.RawMessage

+ type JSONRPCResponse struct

+ Error *RPCError

+ ID *json.RawMessage

+ JSONRPC string

+ Result interface{}

+ type MCPContentBlock struct

+ Text string

+ Type string

+ type MCPServer struct

+ func NewMCPServer(router *ToolRouter, logger *slog.Logger) *MCPServer

+ func (s *MCPServer) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ type MCPToolCallParams struct

+ Arguments map[string]interface{}

+ Name string

+ type MCPToolDef struct

+ Description string

+ InputSchema interface{}

+ Name string

+ type MCPToolResult struct

+ Content []MCPContentBlock

+ IsError bool

+ type ProcessTurnRequest struct

+ IMChannelID string

+ IMUserID string

+ SessionID string

+ UserMessage string

+ WorkspaceID string

+ type RPCError struct

+ Code int

+ Message string

+ type SSEBroker struct

+ func NewSSEBroker() *SSEBroker

+ func (b *SSEBroker) Publish(sessionID string, event *StreamClientEvent)

+ func (b *SSEBroker) Subscribe(sessionID string) *SSESubscriber

+ func (b *SSEBroker) Unsubscribe(sessionID string, sub *SSESubscriber)

+ type SSESubscriber struct

+ Ch chan *StreamClientEvent

+ func (s *SSESubscriber) Close()

+ func (s *SSESubscriber) Done() <-chan struct{}

+ type Server struct

+ func NewServer(cfg Config, store *Store) *Server

+ func (s *Server) CleanupWorker(ctx context.Context, worker *CCWorker)

+ func (s *Server) CreateMCPServer(sessionID, workspaceID, workspaceDir, imChannelID, imUserID string) (*MCPServer, int, func(), error)

+ func (s *Server) Routes() http.Handler

+ func (s *Server) SpawnWorker(ctx context.Context, sessionID, workspaceID, imChannelID, imUserID string) (*CCWorker, error)

+ type Session struct

+ CreatedAt time.Time

+ Epoch int

+ ExternalID *string

+ ID string

+ Source string

+ Status string

+ Title string

+ WorkspaceID string

+ type SessionEvent struct

+ CreatedAt time.Time

+ Ephemeral bool

+ Epoch int

+ EventID string

+ EventType string

+ ID int64

+ Payload json.RawMessage

+ SessionID string

+ Source string

+ type Store struct

+ func NewStore(databaseURL string) (*Store, error)

+ func (s *Store) BumpSessionEpoch(ctx context.Context, id string) (int, error)

+ func (s *Store) Close() error

+ func (s *Store) CreateSession(ctx context.Context, id, workspaceID, title, source string, externalID *string) error

+ func (s *Store) GetEventsSince(ctx context.Context, sessionID string, sinceSeqNum int64, limit int) ([]SessionEvent, error)

+ func (s *Store) GetInternalEventsSince(ctx context.Context, sessionID string, sinceID int64, limit int) ([]SessionEvent, error)

+ func (s *Store) GetSession(ctx context.Context, id string) (*Session, error)

+ func (s *Store) GetSessionEpoch(ctx context.Context, sessionID string) (int, error)

+ func (s *Store) InsertEvents(ctx context.Context, sessionID string, epoch int, events []EventInput) ([]InsertedEvent, error)

+ func (s *Store) InsertInternalEvents(ctx context.Context, sessionID string, events []InternalEventInput) error

+ func (s *Store) UpdateWorkerHeartbeat(ctx context.Context, sessionID string, epoch int) error

+ func (s *Store) UpdateWorkerState(ctx context.Context, sessionID string, epoch int, state string, ...) error

+ func (s *Store) UpsertWorker(ctx context.Context, sessionID string, epoch int) error

+ type StreamClientEvent struct

+ CreatedAt string

+ EventID string

+ EventType string

+ Payload json.RawMessage

+ SequenceNum int64

+ Source string

+ type ToolRouter struct

+ func NewToolRouter(cfg ToolRouterConfig, logger *slog.Logger) *ToolRouter

+ func (r *ToolRouter) Route(ctx context.Context, toolName string, args map[string]interface{}) (*MCPToolResult, error)

+ type ToolRouterConfig struct

+ AgentserverURL string

+ ExecutorRegistryURL string

+ IMBridgeSecret string

+ IMBridgeURL string

+ IMChannelID string

+ IMUserID string

+ SessionID string

+ WorkspaceDir string

+ WorkspaceID string

+ type TurnLock struct

+ func NewTurnLock() *TurnLock

+ func (t *TurnLock) Acquire(sessionID string)

+ func (t *TurnLock) Release(sessionID string)

+ type VikingClient struct

+ func NewVikingClient(baseURL, apiKey string) *VikingClient

+ func (v *VikingClient) CreateFile(ctx context.Context, vikingURI string, content []byte) error

+ func (v *VikingClient) DownloadTree(ctx context.Context, vikingURI, localDir string) error

+ func (v *VikingClient) UploadFile(ctx context.Context, vikingURI, content string) error

+ type Worker struct

+ Epoch int

+ ExternalMetadata json.RawMessage

+ LastHeartbeatAt *time.Time

+ RegisteredAt time.Time

+ RequiresActionDetails json.RawMessage

+ SessionID string

+ State string

+ type WorkerJWTClaims struct

+ Epoch int

+ Exp int64

+ SessionID string

+ WorkspaceID string

+ func ValidateWorkerJWT(secret []byte, token string) (*WorkerJWTClaims, error)

+ type WorkerStateRequest struct

+ ExternalMetadata json.RawMessage

+ RequiresActionDetails json.RawMessage

+ WorkerEpoch int

+ WorkerStatus string