agentIDBytes = 16
)
+// Agent is one agent and its directory. The directory holds config.json, the
+// user's instructions in instructions.md, the chat, and the usage file.
+type Agent struct {
+ dir string
+ config AgentConfig
+}
+
+// AgentConfig holds the settings of the config.json of an agent. ID is random,
+// made once when the file is made, and never changes. The cache key comes from
+// the ID, so a rename keeps the prompt cache and two agents with one name do
+// not share a key. Workdir is the directory where the commands of the agent
+// run. Responsibility and Parent are set together for a child agent.
+type AgentConfig struct {
+ Name string `json:"name"`
+ ID string `json:"id"`
+ Model string `json:"model"`
+ Effort string `json:"effort"`
+ Workdir string `json:"workdir"`
+ Responsibility string `json:"responsibility,omitempty"`
+ Parent string `json:"parent,omitempty"`
+}
+
// openAgent reads the config.json of the agent in dir. When the file does not
// exist, it makes the directory and writes the file with the defaults and a new
// ID. A missing field of an existing file takes the default, but the ID must be
}, nil
}
-// Agent is one agent and its directory. The directory holds config.json, the
-// user's instructions in instructions.md, the chat, and the usage file.
-type Agent struct {
- dir string
- config AgentConfig
-}
-
// Images returns the shared data-directory image store, also for a child.
func (a *Agent) Images() *ImageStore {
root := a.dir
return string(data), nil
}
-// AgentConfig holds the settings of the config.json of an agent. ID is random,
-// made once when the file is made, and never changes. The cache key comes from
-// the ID, so a rename keeps the prompt cache and two agents with one name do
-// not share a key. Workdir is the directory where the commands of the agent
-// run. Responsibility and Parent are set together for a child agent.
-type AgentConfig struct {
- Name string `json:"name"`
- ID string `json:"id"`
- Model string `json:"model"`
- Effort string `json:"effort"`
- Workdir string `json:"workdir"`
- Responsibility string `json:"responsibility,omitempty"`
- Parent string `json:"parent,omitempty"`
-}
-
// validate makes sure that each setting has a valid value.
func (c AgentConfig) validate() error {
if c.Name == "" || strings.ContainsAny(c.Name, "\r\n") {
"strings"
)
+// BackupReader checks cancellation between reads of an archive or file.
+type BackupReader struct {
+ ctx context.Context
+ reader io.Reader
+}
+
+func (r BackupReader) Read(data []byte) (int, error) {
+ if r.ctx.Err() != nil {
+ return 0, r.ctx.Err()
+ }
+
+ count, err := r.reader.Read(data)
+ if err != nil {
+ return count, fmt.Errorf("read backup data: %w", err)
+ }
+
+ return count, nil
+}
+
// backupArchive writes the files in root to path through a private temporary
// file and a rename. Only directories and regular files enter the archive.
func backupArchive(ctx context.Context, root, path string) (size int64, err error) {
return nil
}
-// BackupReader checks cancellation between reads of an archive or file.
-type BackupReader struct {
- ctx context.Context
- reader io.Reader
-}
-
-func (r BackupReader) Read(data []byte) (int, error) {
- if r.ctx.Err() != nil {
- return 0, r.ctx.Err()
- }
-
- count, err := r.reader.Read(data)
- if err != nil {
- return count, fmt.Errorf("read backup data: %w", err)
- }
-
- return count, nil
-}
-
// backupCopyFile copies the size fixed at the snapshot of path. Growth after
// that point is not part of the copy, and a shorter file is an error.
func backupCopyFile(ctx context.Context, out io.Writer, path string, size int64) (err error) {
buildLeafWindow = 8
)
-func newBuildQueue() *BuildQueue {
- return &BuildQueue{
- queued: make(map[NodeKey]bool),
- running: make(map[NodeKey]bool),
- failed: make(map[NodeKey]bool),
- }
-}
-
// BuildQueue holds the nodes that wait for a model build. A leaf joins it when
// its message is too long to be its own node. A merge joins it when both of its
// children are built and their texts do not fit in one node. So the core never
retry []NodeKey
}
+// BuildMergeHeap holds the merges that can start, by the end of their context.
+type BuildMergeHeap []NodeKey
+
+func newBuildQueue() *BuildQueue {
+ return &BuildQueue{
+ queued: make(map[NodeKey]bool),
+ running: make(map[NodeKey]bool),
+ failed: make(map[NodeKey]bool),
+ }
+}
+
// AddLeaf adds the leaf of message id. The IDs must come in increasing order.
func (q *BuildQueue) AddLeaf(id int) {
q.leaves = append(q.leaves, id)
return NodeKey{}, false
}
-// BuildMergeHeap holds the merges that can start, by the end of their context.
-type BuildMergeHeap []NodeKey
-
func (h BuildMergeHeap) Len() int {
return len(h)
}
usagePath string
}
+// DatedMessage is one message with an optional RFC 3339 date.
+type DatedMessage struct {
+ Kind string `json:"kind"`
+ Text string `json:"text"`
+ Date string `json:"date,omitempty"`
+}
+
// Run appends the entries. It waits for the builds only before a user message,
// as a turn does, so up to buildMax builds run at once in between, and it waits
// for all builds at the end. It returns the number of messages appended and why
return r.memory.Wait(ctx)
}
-// DatedMessage is one message with an optional RFC 3339 date.
-type DatedMessage struct {
- Kind string `json:"kind"`
- Text string `json:"text"`
- Date string `json:"date,omitempty"`
-}
-
// buildReadUsage reads the records of usage.jsonl at path, in the order of the
// start of their calls. The file has them in the order of the end of their
// calls, which differs when calls run at once.
images *ImageStore
}
+// ChainFunctionCall holds the fields of a function call output item.
+type ChainFunctionCall struct {
+ Type string `json:"type"`
+ Name string `json:"name"`
+ Arguments string `json:"arguments"`
+ CallID string `json:"call_id"`
+}
+
+// ChainFunctionCallOutput is the input item with the result of a function
+// call.
+type ChainFunctionCallOutput struct {
+ Type string `json:"type"`
+ CallID string `json:"call_id"`
+ Output any `json:"output"`
+}
+
// Run runs the chain from input, the input of its first request, and returns
// the text of the final answer. An error of a log write wraps
// errChainLogWrite.
return input, "", nil
}
-// ChainFunctionCall holds the fields of a function call output item.
-type ChainFunctionCall struct {
- Type string `json:"type"`
- Name string `json:"name"`
- Arguments string `json:"arguments"`
- CallID string `json:"call_id"`
-}
-
-// ChainFunctionCallOutput is the input item with the result of a function
-// call.
-type ChainFunctionCallOutput struct {
- Type string `json:"type"`
- CallID string `json:"call_id"`
- Output any `json:"output"`
-}
-
// chainInputItem returns an output item as an input item for the next request.
// The calls do not store their items, so the item goes back whole, without its
// id.
// must run "onechat login".
var errChatGPTLoginNeeded = errors.New(`login needed: run "onechat login"`)
-// chatgptLogin runs "onechat login": the device code login into the token file
-// of the data directory. It holds the lock of the token file, so it does not
-// run while the daemon runs.
-func chatgptLogin(ctx context.Context, out io.Writer) (err error) {
- dir, err := dataDir()
- if err != nil {
- return err
- }
-
- tokens, err := lockTokens(dir)
- if err != nil {
- return err
- }
-
- defer func() {
- err = errors.Join(err, tokens.Close())
- }()
-
- return newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer).Login(ctx, out)
-}
-
-// newChatGPTAuth returns the authorizer for the token file at path, with the
-// login service at issuer.
-func newChatGPTAuth(path, issuer string) *ChatGPTAuth {
- return &ChatGPTAuth{
- path: path,
- issuer: issuer,
- http: &http.Client{},
- }
-}
-
// ChatGPTAuth authorizes model calls to the ChatGPT backend with the tokens in
// the token file. The tokens belong to OneChat only. The Codex CLI rotates the
// refresh token on each refresh, so two programs with one token file log each
http *http.Client
}
+// ChatGPTDeviceCode is the answer of the user code endpoint. The interval is a
+// number or a string of seconds.
+type ChatGPTDeviceCode struct {
+ DeviceAuthID string `json:"device_auth_id"`
+ UserCode string `json:"user_code"`
+ UserCodeAlt string `json:"usercode"`
+ Interval json.RawMessage `json:"interval"`
+}
+
+// ChatGPTTokenFile is the content of the token file.
+type ChatGPTTokenFile struct {
+ AccessToken string `json:"access_token"`
+ RefreshToken string `json:"refresh_token"`
+ IDToken string `json:"id_token"`
+ AccountID string `json:"account_id"`
+}
+
+// ChatGPTTokenResponse is the answer of the token endpoint.
+type ChatGPTTokenResponse struct {
+ AccessToken string `json:"access_token"`
+ RefreshToken string `json:"refresh_token"`
+ IDToken string `json:"id_token"`
+}
+
+// ChatGPTDeviceGrant is the answer of the device token endpoint after the user
+// approves the code.
+type ChatGPTDeviceGrant struct {
+ AuthorizationCode string `json:"authorization_code"`
+ CodeVerifier string `json:"code_verifier"`
+}
+
+// newChatGPTAuth returns the authorizer for the token file at path, with the
+// login service at issuer.
+func newChatGPTAuth(path, issuer string) *ChatGPTAuth {
+ return &ChatGPTAuth{
+ path: path,
+ issuer: issuer,
+ http: &http.Client{},
+ }
+}
+
// Authorize sets the access token and the account ID, after a refresh when the
// token expires within chatgptRefreshMargin.
func (a *ChatGPTAuth) Authorize(ctx context.Context, header http.Header) error {
return response.StatusCode, nil
}
-// ChatGPTDeviceCode is the answer of the user code endpoint. The interval is a
-// number or a string of seconds.
-type ChatGPTDeviceCode struct {
- DeviceAuthID string `json:"device_auth_id"`
- UserCode string `json:"user_code"`
- UserCodeAlt string `json:"usercode"`
- Interval json.RawMessage `json:"interval"`
-}
+// chatgptLogin runs "onechat login": the device code login into the token file
+// of the data directory. It holds the lock of the token file, so it does not
+// run while the daemon runs.
+func chatgptLogin(ctx context.Context, out io.Writer) (err error) {
+ dir, err := dataDir()
+ if err != nil {
+ return err
+ }
-// ChatGPTTokenFile is the content of the token file.
-type ChatGPTTokenFile struct {
- AccessToken string `json:"access_token"`
- RefreshToken string `json:"refresh_token"`
- IDToken string `json:"id_token"`
- AccountID string `json:"account_id"`
-}
+ tokens, err := lockTokens(dir)
+ if err != nil {
+ return err
+ }
-// ChatGPTTokenResponse is the answer of the token endpoint.
-type ChatGPTTokenResponse struct {
- AccessToken string `json:"access_token"`
- RefreshToken string `json:"refresh_token"`
- IDToken string `json:"id_token"`
-}
+ defer func() {
+ err = errors.Join(err, tokens.Close())
+ }()
-// ChatGPTDeviceGrant is the answer of the device token endpoint after the user
-// approves the code.
-type ChatGPTDeviceGrant struct {
- AuthorizationCode string `json:"authorization_code"`
- CodeVerifier string `json:"code_verifier"`
+ return newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer).Login(ctx, out)
}
// chatgptTokenExpiry returns the expiry of a JWT from its exp claim.
access []string
}
+func newChatGPTFake(t *testing.T) *ChatGPTFake {
+ t.Helper()
+
+ f := &ChatGPTFake{}
+ mux := http.NewServeMux()
+ mux.HandleFunc("POST /api/accounts/deviceauth/usercode", f.userCode)
+ mux.HandleFunc("POST /api/accounts/deviceauth/token", f.deviceToken)
+ mux.HandleFunc("POST /oauth/token", f.oauthToken)
+
+ f.Server = httptest.NewServer(mux)
+ t.Cleanup(f.Close)
+
+ return f
+}
+
+func newTestChatGPTAuth(t *testing.T, server *ChatGPTFake) *ChatGPTAuth {
+ t.Helper()
+
+ return newChatGPTAuth(filepath.Join(t.TempDir(), chatgptTokenFileName), server.URL)
+}
+
// token returns a new access token that expires after ttl, for a token file
// that a test writes.
func (f *ChatGPTFake) token(ttl time.Duration) string {
chatgptWriteJSON(w, answer)
}
-func newChatGPTFake(t *testing.T) *ChatGPTFake {
- t.Helper()
-
- f := &ChatGPTFake{}
- mux := http.NewServeMux()
- mux.HandleFunc("POST /api/accounts/deviceauth/usercode", f.userCode)
- mux.HandleFunc("POST /api/accounts/deviceauth/token", f.deviceToken)
- mux.HandleFunc("POST /oauth/token", f.oauthToken)
-
- f.Server = httptest.NewServer(mux)
- t.Cleanup(f.Close)
-
- return f
-}
-
-func newTestChatGPTAuth(t *testing.T, server *ChatGPTFake) *ChatGPTAuth {
- t.Helper()
-
- return newChatGPTAuth(filepath.Join(t.TempDir(), chatgptTokenFileName), server.URL)
-}
-
func chatgptWriteTestTokens(t *testing.T, auth *ChatGPTAuth, access, refresh string) {
t.Helper()
// bytes, and a sample line as the ruler gets its content copied.
var compactorRuler = strings.Repeat("-", nodeBytes)
-// newCompactor returns the compactor of an agent, which sends its calls through
-// client with model. instructions gives the system prompt of the agent at each
-// build. tools are the tools of that agent, so a build sends the same prompt
-// and tools as its turns.
-func newCompactor(client *ModelClient, instructions func() (string, error), model ModelConfig, tools []ModelTool, images *ImageStore) *Compactor {
- return &Compactor{
- client: client,
- model: model,
- instructions: instructions,
- tools: tools,
- images: images,
- }
-}
-
// Compactor builds the nodes of one agent with a model. A compaction is a call
// like a turn, with the same system prompt and tools, so it reads them from the
// prompt cache of the turns. The tools are never called. After them comes one
images *ImageStore
}
+// newCompactor returns the compactor of an agent, which sends its calls through
+// client with model. instructions gives the system prompt of the agent at each
+// build. tools are the tools of that agent, so a build sends the same prompt
+// and tools as its turns.
+func newCompactor(client *ModelClient, instructions func() (string, error), model ModelConfig, tools []ModelTool, images *ImageStore) *Compactor {
+ return &Compactor{
+ client: client,
+ model: model,
+ instructions: instructions,
+ tools: tools,
+ images: images,
+ }
+}
+
// Summarize returns the line of the node of task.
func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) {
text, err := compactorTask(task)
configEndpointAPI = "api"
)
+// Config holds the settings of config.json in the data directory.
+type Config struct {
+ Endpoint string `json:"endpoint"`
+ Voice VoiceConfig `json:"voice"`
+ Compactor ModelConfig `json:"compactor"`
+}
+
+// ModelConfig names a model and its reasoning effort.
+type ModelConfig struct {
+ Model string `json:"model"`
+ Effort string `json:"effort"`
+}
+
// loadConfig reads config.json in dir over the defaults. A missing file gives
// the defaults. An unknown field is an error, so a typo does not become a
// silent default.
return config, nil
}
-// Config holds the settings of config.json in the data directory.
-type Config struct {
- Endpoint string `json:"endpoint"`
- Voice VoiceConfig `json:"voice"`
- Compactor ModelConfig `json:"compactor"`
-}
-
// validate makes sure that each setting has a valid value.
func (c Config) validate() error {
if c.Endpoint != configEndpointChatGPT && c.Endpoint != configEndpointAPI {
return nil
}
-// ModelConfig names a model and its reasoning effort.
-type ModelConfig struct {
- Model string `json:"model"`
- Effort string `json:"effort"`
-}
-
// defaultConfig returns the settings that apply when config.json does not
// set them.
func defaultConfig() Config {
daemonClientEvents = 1024
)
-// daemon runs "onechat daemon": it owns the root, its children, and their
-// files, and serves the clients on the Unix socket in the data directory until
-// ctx ends. When telegram.json exists, it also runs the Telegram bot. It also
-// keeps a linked WhatsApp device connected. It holds the lock of the token
-// file, so it is the only process that refreshes the tokens and opens the
-// WhatsApp session. Its log lines go to out, through the loggers of
-// daemonLogger.
-func daemon(ctx context.Context, out io.Writer) (err error) {
- dir, err := dataDir()
- if err != nil {
- return err
- }
-
- // Check the socket path before the daemon writes a file, so that a data
- // directory with a long path stays empty.
- socket := filepath.Join(dir, socketFileName)
- err = checkSocketPath(socket)
- if err != nil {
- return err
- }
-
- config, err := loadConfig(dir)
- if err != nil {
- return err
- }
-
- botSettings, hasBot, err := loadTelegramConfig(dir)
- if err != nil {
- return err
- }
-
- tokens, err := lockTokens(dir)
- if err != nil {
- return err
- }
-
- defer func() {
- err = errors.Join(err, tokens.Close())
- }()
-
- whatsappSetDeviceProps()
-
- logger := daemonLogger(out, "onechat: ")
- whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient)
- if err != nil {
- return err
- }
-
- defer func() {
- err = errors.Join(err, whatsapp.Close())
- }()
-
- address, auth, err := modelEndpoint(config, dir)
- if err != nil {
- return err
- }
-
- agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger)
- if err != nil {
- return err
- }
-
- defer func() {
- err = errors.Join(err, agents.Close())
- }()
-
- whatsapp.inbox.voice = &VoiceTranscriber{config: config.Voice}
-
- root := agents[0]
- agent := root.agent
- schedules := root.schedules
-
- listener, err := daemonListenSocket(socket)
- if err != nil {
- return err
- }
-
- linked := "not linked"
- if whatsapp.isLinked() {
- linked = "linked"
- }
-
- logger.Info("ready", "version", version, "socket", listener.Addr(), "model", agent.config.Model, "effort", agent.config.Effort, "workdir", agent.config.Workdir, "whatsapp", linked, "schedules", schedules.Len(), "children", agents.ChildNames())
-
- var telegram *TelegramBot
-
- if hasBot {
- telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: "))
-
- telegram.voice = &VoiceTranscriber{config: config.Voice}
-
- logger.Info("Telegram bot started")
- }
-
- return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
- return root.Loop(whatsapp, notify)
- }, agents[1:]...)
-}
-
// DaemonHub holds the connected clients and sends each event to all of them. It
// runs the WhatsApp commands of the clients on whatsapp. snapshot locks all
// agents and subagents during a backup copy.
workers sync.WaitGroup
}
+// DaemonClient is one connected client. Its events wait in a channel for the
+// writer, so a slow client does not stop the agent.
+type DaemonClient struct {
+ conn net.Conn
+ events chan Event
+ once sync.Once
+}
+
// Accept serves each client of listener until the listener closes. A new
// client gets a ready event with the name of the agent, and a busy event when a
// turn runs.
h.workers.Wait()
}
-// DaemonClient is one connected client. Its events wait in a channel for the
-// writer, so a slow client does not stop the agent.
-type DaemonClient struct {
- conn net.Conn
- events chan Event
- once sync.Once
+// daemon runs "onechat daemon": it owns the root, its children, and their
+// files, and serves the clients on the Unix socket in the data directory until
+// ctx ends. When telegram.json exists, it also runs the Telegram bot. It also
+// keeps a linked WhatsApp device connected. It holds the lock of the token
+// file, so it is the only process that refreshes the tokens and opens the
+// WhatsApp session. Its log lines go to out, through the loggers of
+// daemonLogger.
+func daemon(ctx context.Context, out io.Writer) (err error) {
+ dir, err := dataDir()
+ if err != nil {
+ return err
+ }
+
+ // Check the socket path before the daemon writes a file, so that a data
+ // directory with a long path stays empty.
+ socket := filepath.Join(dir, socketFileName)
+ err = checkSocketPath(socket)
+ if err != nil {
+ return err
+ }
+
+ config, err := loadConfig(dir)
+ if err != nil {
+ return err
+ }
+
+ botSettings, hasBot, err := loadTelegramConfig(dir)
+ if err != nil {
+ return err
+ }
+
+ tokens, err := lockTokens(dir)
+ if err != nil {
+ return err
+ }
+
+ defer func() {
+ err = errors.Join(err, tokens.Close())
+ }()
+
+ whatsappSetDeviceProps()
+
+ logger := daemonLogger(out, "onechat: ")
+ whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient)
+ if err != nil {
+ return err
+ }
+
+ defer func() {
+ err = errors.Join(err, whatsapp.Close())
+ }()
+
+ address, auth, err := modelEndpoint(config, dir)
+ if err != nil {
+ return err
+ }
+
+ agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger)
+ if err != nil {
+ return err
+ }
+
+ defer func() {
+ err = errors.Join(err, agents.Close())
+ }()
+
+ whatsapp.inbox.voice = &VoiceTranscriber{config: config.Voice}
+
+ root := agents[0]
+ agent := root.agent
+ schedules := root.schedules
+
+ listener, err := daemonListenSocket(socket)
+ if err != nil {
+ return err
+ }
+
+ linked := "not linked"
+ if whatsapp.isLinked() {
+ linked = "linked"
+ }
+
+ logger.Info("ready", "version", version, "socket", listener.Addr(), "model", agent.config.Model, "effort", agent.config.Effort, "workdir", agent.config.Workdir, "whatsapp", linked, "schedules", schedules.Len(), "children", agents.ChildNames())
+
+ var telegram *TelegramBot
+
+ if hasBot {
+ telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: "))
+
+ telegram.voice = &VoiceTranscriber{config: config.Voice}
+
+ logger.Info("Telegram bot started")
+ }
+
+ return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
+ return root.Loop(whatsapp, notify)
+ }, agents[1:]...)
}
// daemonListenSocket listens on the Unix socket at path, with mode 0600. The
"time"
)
+// DaemonAgent owns the open files and model client of one persistent agent. Its
+// loop must stop before Close releases the files.
+type DaemonAgent struct {
+ agent *Agent
+ memory *Memory
+ client *ModelClient
+ subagents *Subagents
+ schedules *Schedules
+ clock func() time.Time
+ logger *slog.Logger
+ openChild func(context.Context, *Agent) (*DaemonAgent, error)
+}
+
+// DaemonAgents holds the root first and then its children in directory order.
+type DaemonAgents []*DaemonAgent
+
// openDaemonAgents opens the root and its direct children. An invalid child
// stops startup, and each opened agent is closed when startup fails.
func openDaemonAgents(ctx context.Context, dir, address string, auth ModelAuthorizer, compactor ModelConfig, clock func() time.Time, logger *slog.Logger) (agents DaemonAgents, err error) {
}, nil
}
-// DaemonAgent owns the open files and model client of one persistent agent. Its
-// loop must stop before Close releases the files.
-type DaemonAgent struct {
- agent *Agent
- memory *Memory
- client *ModelClient
- subagents *Subagents
- schedules *Schedules
- clock func() time.Time
- logger *slog.Logger
- openChild func(context.Context, *Agent) (*DaemonAgent, error)
-}
-
// Loop makes the turn loop with its event destination. Only the root has a
// WhatsApp link. Child events stay with their own logs.
func (a *DaemonAgent) Loop(whatsapp *WhatsApp, notify func(Event)) (*TurnLoop, error) {
return errors.Join(a.subagents.Close(), a.memory.Close())
}
-// DaemonAgents holds the root first and then its children in directory order.
-type DaemonAgents []*DaemonAgent
-
// Close closes all agents, including after a partial startup.
func (a DaemonAgents) Close() error {
var err error
"time"
)
-// daemonLogger returns a logger for one daemon part. JOURNAL_STREAM selects
-// syslog priorities without a time stamp. Other output carries the standard
-// time stamp and level.
-func daemonLogger(out io.Writer, prefix string) *slog.Logger {
- return slog.New(&DaemonLogHandler{
- out: out,
- prefix: prefix,
- journal: os.Getenv("JOURNAL_STREAM") != "",
- mutex: &sync.Mutex{},
- })
-}
-
// DaemonLogHandler writes short event lines with ordered attributes. Copies
// share the writer lock. The err attribute follows all other attributes as a
// cause. Journal lines carry a syslog priority instead of a time and level.
return &clone
}
+// daemonLogger returns a logger for one daemon part. JOURNAL_STREAM selects
+// syslog priorities without a time stamp. Other output carries the standard
+// time stamp and level.
+func daemonLogger(out io.Writer, prefix string) *slog.Logger {
+ return slog.New(&DaemonLogHandler{
+ out: out,
+ prefix: prefix,
+ journal: os.Getenv("JOURNAL_STREAM") != "",
+ mutex: &sync.Mutex{},
+ })
+}
+
// daemonLogAttr resolves and writes one attribute, or keeps its cause for the
// end of the line. Groups flatten into dotted keys and empty attributes produce
// no text.
"sync"
)
+// DaemonMessages owns the durable queue between the root and its children and
+// the routed inputs. The mutex serializes queue writes and delivery. Backups
+// take it before the locks of the loops, memory, and subagents. A stored log ID
+// makes delivery idempotent, including a stop between the log append and the
+// wake.
+type DaemonMessages struct {
+ mutex sync.Mutex
+ path string
+ loops map[string]*TurnLoop
+ ordered []*TurnLoop
+ state DaemonMessageFile
+ failure error
+ routes *Routes
+}
+
+// DaemonMessageFile holds the sequence, unacknowledged deliveries, and receipts
+// for WhatsApp batches until the inbox removes them. A receipt stays when its
+// turn finishes before inbox removal.
+type DaemonMessageFile struct {
+ Next int `json:"next"`
+ Pending []DaemonMessage `json:"pending"`
+ WhatsAppBatches []int `json:"whatsapp_batches,omitempty"`
+}
+
+// DaemonMessage is one queued delivery. LogID is set before the log append. A
+// nil LogID means that delivery has not started. WhatsAppBatch marks a routed
+// external batch, whose event source is WhatsApp rather than its queue sender.
+type DaemonMessage struct {
+ ID int `json:"id"`
+ Source string `json:"source"`
+ Target string `json:"target"`
+ Kind string `json:"kind"`
+ Text string `json:"text"`
+ LogID *int `json:"log_id,omitempty"`
+ WhatsAppBatch int `json:"whatsapp_batch,omitempty"`
+}
+
// openDaemonMessages connects the root and its children. It restores queued
// messages before the loops start. A queue entry stays until its recipient
// finishes the turn that takes it, so a stop can interrupt delivery or a turn.
return m, nil
}
-// DaemonMessages owns the durable queue between the root and its children and
-// the routed inputs. The mutex serializes queue writes and delivery. Backups
-// take it before the locks of the loops, memory, and subagents. A stored log ID
-// makes delivery idempotent, including a stop between the log append and the
-// wake.
-type DaemonMessages struct {
- mutex sync.Mutex
- path string
- loops map[string]*TurnLoop
- ordered []*TurnLoop
- state DaemonMessageFile
- failure error
- routes *Routes
-}
-
// deliver records one queued message and wakes its recipient. The caller holds
// the mutex, or owns the queue before the loops start.
func (m *DaemonMessages) deliver(message *DaemonMessage) error {
return nil
}
-
-// DaemonMessageFile holds the sequence, unacknowledged deliveries, and receipts
-// for WhatsApp batches until the inbox removes them. A receipt stays when its
-// turn finishes before inbox removal.
-type DaemonMessageFile struct {
- Next int `json:"next"`
- Pending []DaemonMessage `json:"pending"`
- WhatsAppBatches []int `json:"whatsapp_batches,omitempty"`
-}
-
-// DaemonMessage is one queued delivery. LogID is set before the log append. A
-// nil LogID means that delivery has not started. WhatsAppBatch marks a routed
-// external batch, whose event source is WhatsApp rather than its queue sender.
-type DaemonMessage struct {
- ID int `json:"id"`
- Source string `json:"source"`
- Target string `json:"target"`
- Kind string `json:"kind"`
- Text string `json:"text"`
- LogID *int `json:"log_id,omitempty"`
- WhatsAppBatch int `json:"whatsapp_batch,omitempty"`
-}
done []chan error
}
+func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest {
+ t.Helper()
+
+ _, err := os.Stat(filepath.Join(dir, daemonAgentDirName, "children"))
+ if errors.Is(err, os.ErrNotExist) {
+ daemonChildConfig(t, dir, "Garden")
+ daemonChildConfig(t, dir, "Library")
+ } else {
+ require.NoError(t, err)
+ }
+
+ test := &DaemonMessageTest{
+ logs: &SyncBuffer{},
+ }
+ test.agents, err = openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, daemonTestLogger(test.logs))
+ require.NoError(t, err)
+ t.Cleanup(func() {
+ test.Close(t)
+ })
+
+ for _, agent := range test.agents {
+ events := &TurnEventRecorder{}
+ loop, err := agent.Loop(nil, events.Add)
+ require.NoError(t, err)
+
+ test.events = append(test.events, events)
+ test.loops = append(test.loops, loop)
+ }
+
+ test.messages, err = openDaemonMessages(test.loops)
+ require.NoError(t, err)
+
+ return test
+}
+
func (r *DaemonMessageTest) Start(t *testing.T, indices ...int) {
t.Helper()
r.agents = nil
}
-
-func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest {
- t.Helper()
-
- _, err := os.Stat(filepath.Join(dir, daemonAgentDirName, "children"))
- if errors.Is(err, os.ErrNotExist) {
- daemonChildConfig(t, dir, "Garden")
- daemonChildConfig(t, dir, "Library")
- } else {
- require.NoError(t, err)
- }
-
- test := &DaemonMessageTest{
- logs: &SyncBuffer{},
- }
- test.agents, err = openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, daemonTestLogger(test.logs))
- require.NoError(t, err)
- t.Cleanup(func() {
- test.Close(t)
- })
-
- for _, agent := range test.agents {
- events := &TurnEventRecorder{}
- loop, err := agent.Loop(nil, events.Add)
- require.NoError(t, err)
-
- test.events = append(test.events, events)
- test.loops = append(test.loops, loop)
- }
-
- test.messages, err = openDaemonMessages(test.loops)
- require.NoError(t, err)
-
- return test
-}
"note": true,
}
+// Log is the append-only message log of one chat. It writes one JSON line for
+// each message to main/YYYY-MM-DD.jsonl, named for the local day of the
+// message. It never edits or deletes a line. Each line goes to the operating
+// system before Append returns, so a crash of the process loses no message. An
+// exclusive lock on the lock file makes sure that only one process writes the
+// log. After a failed write, the end of the file is not known, so the log
+// refuses all writes until it is opened again. The caller serializes all calls.
+//
+// The log keeps no message in memory. The index index/main.idx holds the
+// position of message i in record i, so a read of a message is one read of the
+// index and one of its day file. At open the log indexes the lines after the
+// last indexed one, so a missing index is built again from the day files.
+type Log struct {
+ clock func() time.Time
+ lock *os.File
+ journal Journal
+ index *Records
+ failure error
+}
+
+// Message is one entry of the chat log. ID is its position in the log, from 0.
+// A rebuild assigns positions in the merged log. Size is the UTF-8 byte length
+// of Text. Date is the message time in RFC 3339 format, supplied at append or
+// taken from the clock.
+type Message struct {
+ ID int `json:"i"`
+ Kind string `json:"kind"`
+ Text string `json:"text"`
+ Size int `json:"size"`
+ Date string `json:"date"`
+}
+
// openLog locks the chat in root and indexes its new lines. The clock gives the
// date of each new message.
func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {
return l, nil
}
-// Log is the append-only message log of one chat. It writes one JSON line for
-// each message to main/YYYY-MM-DD.jsonl, named for the local day of the
-// message. It never edits or deletes a line. Each line goes to the operating
-// system before Append returns, so a crash of the process loses no message. An
-// exclusive lock on the lock file makes sure that only one process writes the
-// log. After a failed write, the end of the file is not known, so the log
-// refuses all writes until it is opened again. The caller serializes all calls.
-//
-// The log keeps no message in memory. The index index/main.idx holds the
-// position of message i in record i, so a read of a message is one read of the
-// index and one of its day file. At open the log indexes the lines after the
-// last indexed one, so a missing index is built again from the day files.
-type Log struct {
- clock func() time.Time
- lock *os.File
- journal Journal
- index *Records
- failure error
-}
-
// Close releases the lock. The log refuses all calls after Close.
func (l *Log) Close() error {
err := l.journal.Close()
return l.index.Append(record)
}
-// Message is one entry of the chat log. ID is its position in the log, from 0.
-// A rebuild assigns positions in the merged log. Size is the UTF-8 byte length
-// of Text. Date is the message time in RFC 3339 format, supplied at append or
-// taken from the clock.
-type Message struct {
- ID int `json:"i"`
- Kind string `json:"kind"`
- Text string `json:"text"`
- Size int `json:"size"`
- Date string `json:"date"`
-}
-
// logValidateMessage makes sure that a message has a known kind, valid UTF-8
// text, the size of its text, and a valid date.
func logValidateMessage(message Message) error {
// positionBytes is the size of an encoded position.
const positionBytes = 16
+// Records is a file of fixed-size Records, numbered from 0. Record n starts at
+// byte n*size. A record that was never written reads as zeros. An index is data
+// that the day files can give again, so Records cuts a partial record at the
+// end of the file.
+type Records struct {
+ file *os.File
+ size int
+ count int
+}
+
// openRecords opens or creates the record file at path with records of size
// bytes.
func openRecords(path string, size int) (*Records, error) {
return r, nil
}
-// encodePosition writes place into the first positionBytes bytes of record.
-func encodePosition(record []byte, place Position) {
- binary.LittleEndian.PutUint32(record[0:], place.day)
- binary.LittleEndian.PutUint32(record[4:], place.length)
- binary.LittleEndian.PutUint64(record[8:], place.offset)
-}
-
-// decodePosition reads a position from the first positionBytes bytes of record.
-func decodePosition(record []byte) Position {
- return Position{
- day: binary.LittleEndian.Uint32(record[0:]),
- length: binary.LittleEndian.Uint32(record[4:]),
- offset: binary.LittleEndian.Uint64(record[8:]),
- }
-}
-
-// Records is a file of fixed-size Records, numbered from 0. Record n starts at
-// byte n*size. A record that was never written reads as zeros. An index is data
-// that the day files can give again, so Records cuts a partial record at the
-// end of the file.
-type Records struct {
- file *os.File
- size int
- count int
-}
-
// Read returns record n, or false when the file does not reach it.
func (r *Records) Read(n int) ([]byte, bool, error) {
if n >= r.count {
func (r *Records) Close() error {
return r.file.Close()
}
+
+// encodePosition writes place into the first positionBytes bytes of record.
+func encodePosition(record []byte, place Position) {
+ binary.LittleEndian.PutUint32(record[0:], place.day)
+ binary.LittleEndian.PutUint32(record[4:], place.length)
+ binary.LittleEndian.PutUint64(record[8:], place.offset)
+}
+
+// decodePosition reads a position from the first positionBytes bytes of record.
+func decodePosition(record []byte) Position {
+ return Position{
+ day: binary.LittleEndian.Uint32(record[0:]),
+ length: binary.LittleEndian.Uint32(record[4:]),
+ offset: binary.LittleEndian.Uint64(record[8:]),
+ }
+}
"go.mau.fi/whatsmeow/types"
)
+// Routes owns the root's source map. Writes replace the private file before
+// changing the map in memory. The mutex protects reads and writes from tools
+// and status commands. Logs omit source IDs, which can hold phone numbers.
+type Routes struct {
+ mutex sync.Mutex
+ path string
+ logger *slog.Logger
+ targets map[string]string
+}
+
// openRoutes reads the source-to-child map. A missing file has no routes.
func openRoutes(path string, logger *slog.Logger) (*Routes, error) {
r := &Routes{
return r, nil
}
-// Routes owns the root's source map. Writes replace the private file before
-// changing the map in memory. The mutex protects reads and writes from tools
-// and status commands. Logs omit source IDs, which can hold phone numbers.
-type Routes struct {
- mutex sync.Mutex
- path string
- logger *slog.Logger
- targets map[string]string
-}
-
// Set maps source to a child name, or removes it when target is empty. The
// caller checks that a nonempty target is a child of the root.
func (r *Routes) Set(source, target string) error {
scheduleLateLine = "This run is late: the daemon was stopped at that time."
)
+// Schedules holds the schedules of the agent: tasks that start a turn at one
+// time or at the times of a cron expression. Each change writes the file whole
+// before it returns, so a schedule survives a restart. The mutex protects the
+// schedules, because the tools change them from a turn while Run reads them.
+// changed tells Run that a set or a remove changed the times.
+type Schedules struct {
+ path string
+ owner string
+ clock func() time.Time
+ logger *slog.Logger
+ changed chan struct{}
+
+ mutex sync.Mutex
+ file ScheduleFile
+}
+
+// ScheduleFile is the content of the file of the schedules: the ID of the last
+// schedule, so that an ID is never used again, and the schedules.
+type ScheduleFile struct {
+ Next int `json:"next"`
+ Schedules []Schedule `json:"schedules"`
+}
+
+// Schedule is one schedule: its owner, its ID, the task for the agent, the cron
+// expression of a repeating schedule, empty for a one-time schedule, and its
+// next time.
+type Schedule struct {
+ Owner string `json:"owner"`
+ ID int `json:"id"`
+ Text string `json:"text"`
+ Cron string `json:"cron"`
+ Due time.Time `json:"due"`
+}
+
// openSchedules reads the schedules of the file at path. A missing file holds
// none. The owner names its agent. Schedules without an owner take this name. A
// different stored owner is an error. The clock gives the time and the time
return s, nil
}
-// Schedules holds the schedules of the agent: tasks that start a turn at one
-// time or at the times of a cron expression. Each change writes the file whole
-// before it returns, so a schedule survives a restart. The mutex protects the
-// schedules, because the tools change them from a turn while Run reads them.
-// changed tells Run that a set or a remove changed the times.
-type Schedules struct {
- path string
- owner string
- clock func() time.Time
- logger *slog.Logger
- changed chan struct{}
-
- mutex sync.Mutex
- file ScheduleFile
-}
-
// Run starts the runs of the due schedules in l until ctx ends. It waits until
// the first due time, at most scheduleClockWaitTimeout at a time, and also
// wakes at a set or a remove. So a schedule that came due while the daemon was
return nil
}
-// ScheduleFile is the content of the file of the schedules: the ID of the last
-// schedule, so that an ID is never used again, and the schedules.
-type ScheduleFile struct {
- Next int `json:"next"`
- Schedules []Schedule `json:"schedules"`
-}
-
-// Schedule is one schedule: its owner, its ID, the task for the agent, the cron
-// expression of a repeating schedule, empty for a one-time schedule, and its
-// next time.
-type Schedule struct {
- Owner string `json:"owner"`
- ID int `json:"id"`
- Text string `json:"text"`
- Cron string `json:"cron"`
- Due time.Time `json:"due"`
-}
-
// Message returns the text of a run at now for the log, with the times in the
// location of now. The first line starts with [Schedule ID] and gives the due
// time and once or the cron expression. Then come the text, the line about a
},
}
-// scheduleParseCron parses a cron expression of five fields: minute, hour, day
-// of month, month, and day of week, where 0 and 7 are Sunday. A field is a
-// comma list of items, and an item is *, N, or N-M, where * and N-M can have a
-// step /S. Names, macros, the characters ? L W #, and N/S are errors.
-//
-// It refuses an expression whose runs can be less than scheduleCronGap minutes
-// apart. The check takes the wall times of one day from the hour and the minute
-// fields, and the gap across midnight as if the next day matches too. It also
-// refuses an expression that matches no day in scheduleCronYears years, such as
-// one on February 30.
-func scheduleParseCron(expression string) (ScheduleCron, error) {
- fields := strings.Fields(expression)
- if len(fields) != len(scheduleCronFields) {
- return ScheduleCron{}, fmt.Errorf("the cron expression %q has %d fields, not 5: minute, hour, day of month, month, day of week", expression, len(fields))
- }
-
- var sets [5]uint64
-
- for index, field := range fields {
- set, err := scheduleCronField(field, scheduleCronFields[index].low, scheduleCronFields[index].high)
- if err != nil {
- return ScheduleCron{}, fmt.Errorf("the %s field of %q: %w", scheduleCronFields[index].name, expression, err)
- }
-
- sets[index] = set
- }
-
- c := ScheduleCron{
- minutes: sets[0],
- hours: sets[1],
- days: sets[2],
- months: sets[3],
- weekdays: sets[4],
- anyDay: strings.HasPrefix(fields[2], "*"),
- anyWeekday: strings.HasPrefix(fields[4], "*"),
- description: strings.Join(fields, " "),
- }
-
- if c.weekdays&(1<<7) != 0 {
- c.weekdays |= 1
- }
-
- if c.gap() < scheduleCronGap {
- return ScheduleCron{}, fmt.Errorf("the runs of %q can be less than %d minutes apart: the runs of a schedule must be at least %d minutes apart", expression, scheduleCronGap, scheduleCronGap)
- }
-
- start := time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)
- _, found := c.Next(start)
- if !found {
- return ScheduleCron{}, fmt.Errorf("the cron expression %q matches no day in %d years", expression, scheduleCronYears)
- }
-
- return c, nil
-}
-
// ScheduleCron is a parsed cron expression. Each set holds a bit for each value
// of its field. anyDay and anyWeekday tell that the day field or the weekday
// field starts with *. When neither does, a day matches when either field
return gap
}
+// scheduleParseCron parses a cron expression of five fields: minute, hour, day
+// of month, month, and day of week, where 0 and 7 are Sunday. A field is a
+// comma list of items, and an item is *, N, or N-M, where * and N-M can have a
+// step /S. Names, macros, the characters ? L W #, and N/S are errors.
+//
+// It refuses an expression whose runs can be less than scheduleCronGap minutes
+// apart. The check takes the wall times of one day from the hour and the minute
+// fields, and the gap across midnight as if the next day matches too. It also
+// refuses an expression that matches no day in scheduleCronYears years, such as
+// one on February 30.
+func scheduleParseCron(expression string) (ScheduleCron, error) {
+ fields := strings.Fields(expression)
+ if len(fields) != len(scheduleCronFields) {
+ return ScheduleCron{}, fmt.Errorf("the cron expression %q has %d fields, not 5: minute, hour, day of month, month, day of week", expression, len(fields))
+ }
+
+ var sets [5]uint64
+
+ for index, field := range fields {
+ set, err := scheduleCronField(field, scheduleCronFields[index].low, scheduleCronFields[index].high)
+ if err != nil {
+ return ScheduleCron{}, fmt.Errorf("the %s field of %q: %w", scheduleCronFields[index].name, expression, err)
+ }
+
+ sets[index] = set
+ }
+
+ c := ScheduleCron{
+ minutes: sets[0],
+ hours: sets[1],
+ days: sets[2],
+ months: sets[3],
+ weekdays: sets[4],
+ anyDay: strings.HasPrefix(fields[2], "*"),
+ anyWeekday: strings.HasPrefix(fields[4], "*"),
+ description: strings.Join(fields, " "),
+ }
+
+ if c.weekdays&(1<<7) != 0 {
+ c.weekdays |= 1
+ }
+
+ if c.gap() < scheduleCronGap {
+ return ScheduleCron{}, fmt.Errorf("the runs of %q can be less than %d minutes apart: the runs of a schedule must be at least %d minutes apart", expression, scheduleCronGap, scheduleCronGap)
+ }
+
+ start := time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)
+ _, found := c.Next(start)
+ if !found {
+ return ScheduleCron{}, fmt.Errorf("the cron expression %q matches no day in %d years", expression, scheduleCronYears)
+ }
+
+ return c, nil
+}
+
// scheduleWallTime returns the time of a wall time in location. A wall time
// that the clock skips, at the start of daylight saving time, is later by the
// length of the skip: 02:30 becomes 03:30 when the clock goes from 02:00 to
terminalSource = "terminal"
)
-// checkSocketPath makes sure that path fits in the address of a Unix socket.
-func checkSocketPath(path string) error {
- if len(path) > socketPathBytes {
- return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes)
- }
-
- return nil
-}
-
// SocketRequest is one line from a client: "send" with the text of a user
// message, "interrupt", "stop" with the name of a subagent in Text, or
// "whatsapp" with the text after /whatsapp of a WhatsApp command, or "backup"
Text string `json:"text,omitempty"`
Path string `json:"path,omitempty"`
}
+
+// checkSocketPath makes sure that path fits in the address of a Unix socket.
+func checkSocketPath(path string) error {
+ if len(path) > socketPathBytes {
+ return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes)
+ }
+
+ return nil
+}
projects / onechat / commitdiff