now time.Time
}
-func (c *TestClock) Now() time.Time {
- return c.now
-}
-
func newTestClock() *TestClock {
return &TestClock{
now: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC),
}
}
+
+func (c *TestClock) Now() time.Time {
+ return c.now
+}
errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message")
)
+// Summarizer builds the text of one node with a model.
+type Summarizer interface {
+ Summarize(ctx context.Context, task Task) (string, error)
+}
+
+// Memory is the memory of one chat: the message log, the summary tree, the chat
+// view, the compaction view, and the queue of the builds. The compaction view
+// is the chat view merged further, at the limits of 16 to 32 KB. It takes the
+// same new lines, and it merges again in one batch when it is too large or when
+// the chat view merges. So the compactions share its prefix in the prompt
+// cache. It saves the views to view.json after each change and loads them at
+// start. It never builds the views again from the log, because a new view
+// differs from the live one and so loses the prompt cache.
+//
+// Each write goes to the operating system at once, and a syncer makes the
+// writes durable by group commit in the background. Flush waits until the
+// writes so far are durable.
+//
+// The summarizer builds the nodes in the queue in the background. The start
+// reads the indexes once to fill the queue again. After that, only a new
+// message or a new node adds work. Memory keeps the nodes of the views and the
+// tasks of the running builds. It reads all other messages and nodes from their
+// day files through the indexes.
+type Memory struct {
+ mutex sync.Mutex
+ root string
+ log *Log
+ syncer *Syncer
+ tree *Tree
+ chat *View
+ cview *View
+ queue *BuildQueue
+ summarizer Summarizer
+ logger *slog.Logger
+ ctx context.Context
+ cancel context.CancelFunc
+ workers sync.WaitGroup
+ changed chan struct{}
+}
+
+// Task is the build of the node (Level, Index). Context is the compaction view
+// for the build. Input holds the source "kind: text" of a leaf, or the two
+// child lines of a merge with their newlines as spaces. Images holds the leaf
+// image filenames and is empty for merges.
+type Task struct {
+ Level int
+ Index int
+ Context string
+ Input []string
+ Images []string
+}
+
+// MemoryViewFile is the content of view.json.
+type MemoryViewFile struct {
+ Chat ViewState `json:"chat"`
+ Compaction ViewState `json:"compaction"`
+}
+
// openMemory opens the chat in root and starts the builds that are ready. The
// clock gives the dates of new messages and the days of new nodes. logger gets
// one line for each failed build. The builds stop when ctx ends or at Close.
return m, nil
}
-// Summarizer builds the text of one node with a model.
-type Summarizer interface {
- Summarize(ctx context.Context, task Task) (string, error)
-}
-
-// Memory is the memory of one chat: the message log, the summary tree, the chat
-// view, the compaction view, and the queue of the builds. The compaction view
-// is the chat view merged further, at the limits of 16 to 32 KB. It takes the
-// same new lines, and it merges again in one batch when it is too large or when
-// the chat view merges. So the compactions share its prefix in the prompt
-// cache. It saves the views to view.json after each change and loads them at
-// start. It never builds the views again from the log, because a new view
-// differs from the live one and so loses the prompt cache.
-//
-// Each write goes to the operating system at once, and a syncer makes the
-// writes durable by group commit in the background. Flush waits until the
-// writes so far are durable.
-//
-// The summarizer builds the nodes in the queue in the background. The start
-// reads the indexes once to fill the queue again. After that, only a new
-// message or a new node adds work. Memory keeps the nodes of the views and the
-// tasks of the running builds. It reads all other messages and nodes from their
-// day files through the indexes.
-type Memory struct {
- mutex sync.Mutex
- root string
- log *Log
- syncer *Syncer
- tree *Tree
- chat *View
- cview *View
- queue *BuildQueue
- summarizer Summarizer
- logger *slog.Logger
- ctx context.Context
- cancel context.CancelFunc
- workers sync.WaitGroup
- changed chan struct{}
-}
-
// Close stops the builds, waits for them, makes the writes durable, closes the
// files of the chat, and releases its lock.
func (m *Memory) Close() error {
return !free, nil
}
-
-// Task is the build of the node (Level, Index). Context is the compaction view
-// for the build. Input holds the source "kind: text" of a leaf, or the two
-// child lines of a merge with their newlines as spaces. Images holds the leaf
-// image filenames and is empty for merges.
-type Task struct {
- Level int
- Index int
- Context string
- Input []string
- Images []string
-}
-
-// MemoryViewFile is the content of view.json.
-type MemoryViewFile struct {
- Chat ViewState `json:"chat"`
- Compaction ViewState `json:"compaction"`
-}
// errModelCallLimit tells that a client sent its limit of calls.
var errModelCallLimit = errors.New("call limit reached")
-// modelEndpoint returns the URL and the authorizer of the endpoint in config.
-// The ChatGPT backend uses the token file in dir. The OpenAI API uses the key
-// in OPENAI_API_KEY, and there is no fallback from one to the other. All
-// clients share the authorizer, so one refresh runs at a time.
-func modelEndpoint(config Config, dir string) (string, ModelAuthorizer, error) {
- if config.Endpoint == configEndpointAPI {
- key := os.Getenv("OPENAI_API_KEY")
- if key == "" {
- return "", nil, errors.New("endpoint api needs OPENAI_API_KEY")
- }
-
- return modelAPIURL, ModelAPIKey(key), nil
- }
-
- return modelChatGPTURL, newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer), nil
-}
-
-// newModelClient returns a client for the Responses endpoint at url. It records
-// each call in the usage file at usagePath.
-func newModelClient(url string, auth ModelAuthorizer, cacheKey, usagePath string) *ModelClient {
- return &ModelClient{
- url: url,
- auth: auth,
- cacheKey: cacheKey,
- usage: &ModelUsageLog{
- path: usagePath,
- },
- http: &http.Client{},
- }
-}
-
-// modelUserMessage returns a user input item with one text part for each text.
-func modelUserMessage(texts ...string) any {
- content := make([]ModelInputPart, len(texts))
- for n, text := range texts {
- content[n] = ModelInputPart{
- Type: "input_text",
- Text: text,
- }
- }
-
- return modelUserParts(content)
-}
-
-// modelUserParts keeps text and image parts in one user message.
-func modelUserParts(content []ModelInputPart) any {
- return struct {
- Role string `json:"role"`
- Content []ModelInputPart `json:"content"`
- }{
- Role: "user",
- Content: content,
- }
-}
-
-// modelOutputText returns the text of the message items of output.
-func modelOutputText(output []json.RawMessage) (string, error) {
- var text strings.Builder
-
- for _, raw := range output {
- var item ModelOutputItem
-
- err := json.Unmarshal(raw, &item)
- if err != nil {
- return "", fmt.Errorf("decode output item: %w", err)
- }
-
- if item.Type != "message" {
- continue
- }
-
- for _, part := range item.Content {
- if part.Type == "output_text" {
- text.WriteString(part.Text)
- }
- }
- }
-
- return text.String(), nil
-}
-
// ModelAuthorizer adds the credentials of a model call to the headers of its
// request.
type ModelAuthorizer interface {
// ModelAPIKey authorizes calls to the OpenAI API with a key.
type ModelAPIKey string
-// Authorize sets the key as the bearer token.
-func (k ModelAPIKey) Authorize(ctx context.Context, header http.Header) error {
- header.Set("Authorization", "Bearer "+string(k))
-
- return ctx.Err()
-}
-
// ModelClient sends model calls to a Responses endpoint for one agent. The
// ChatGPT backend and the OpenAI API take the same request body and differ only
// in the URL and the credentials. Every call sends the fixed cache key of the
calls atomic.Int64
}
+// ModelUsageLog appends one JSON line for each model call to usage.jsonl. Calls
+// can run at once, so a mutex serializes the writes.
+type ModelUsageLog struct {
+ mutex sync.Mutex
+ path string
+}
+
+// ModelOutputItem holds the fields of an output item that give its text.
+type ModelOutputItem struct {
+ Type string `json:"type"`
+ Content []struct {
+ Type string `json:"type"`
+ Text string `json:"text"`
+ } `json:"content"`
+}
+
+// ModelInputPart is a text or image part in a user message or tool result.
+type ModelInputPart struct {
+ Type string `json:"type"`
+ Text string `json:"text,omitempty"`
+ ImageURL string `json:"image_url,omitempty"`
+}
+
+// ModelRequest is one model call. Input holds the input items in order, such as
+// modelUserMessage values or output items of an earlier response. Node names
+// the node that a compaction builds, or is nil. Turn names the request of a
+// turn: first or next. Agent names the subagent that makes the call, or is
+// empty for a call of the agent itself. Node, Turn, and Agent go to the usage
+// file, never to the endpoint as fields. Agent also gives the call the cache
+// key of its subagent.
+type ModelRequest struct {
+ Model string
+ Effort string
+ Instructions string
+ Tools []ModelTool
+ Input []any
+ Node *NodeKey
+ Turn string
+ Agent string
+}
+
+// ModelResponse is the result of a model call. Output holds the output items as
+// the endpoint sent them. Text is the text of the message items.
+type ModelResponse struct {
+ Output []json.RawMessage
+ Text string
+ Usage ModelUsage
+}
+
+// ModelRequestBody is the JSON body of a call.
+type ModelRequestBody struct {
+ Model string `json:"model"`
+ Instructions string `json:"instructions"`
+ Input []any `json:"input"`
+ Tools []ModelTool `json:"tools"`
+ Reasoning ModelReasoning `json:"reasoning"`
+ Store bool `json:"store"`
+ Stream bool `json:"stream"`
+ Include []string `json:"include"`
+ PromptCacheKey string `json:"prompt_cache_key"`
+}
+
+type ModelReasoning struct {
+ Effort string `json:"effort"`
+}
+
+// ModelUsageRecord is one line of usage.jsonl. Date is the start of the call,
+// and Seconds its length. Node is the node [l, i] of a compaction. Status is
+// the HTTP status, or 0 when no response came. TextBytes is the size of the
+// text of the reply. Turn names the request of a turn. Agent names the subagent
+// of the call. Quota holds the x-codex-* headers of the response, without the
+// turn state.
+type ModelUsageRecord struct {
+ Date string `json:"date"`
+ Seconds float64 `json:"seconds"`
+ Model string `json:"model"`
+ Node *[2]int `json:"node,omitempty"`
+ Turn string `json:"turn,omitempty"`
+ Agent string `json:"agent,omitempty"`
+ Status int `json:"status"`
+ InputTokens int `json:"input_tokens"`
+ CachedTokens int `json:"cached_tokens"`
+ OutputTokens int `json:"output_tokens"`
+ TextBytes int `json:"text_bytes"`
+ Quota map[string]string `json:"quota,omitempty"`
+ Error string `json:"error,omitempty"`
+}
+
+// ModelTool is the definition of a function tool.
+type ModelTool struct {
+ Type string `json:"type"`
+ Name string `json:"name"`
+ Description string `json:"description"`
+ Parameters any `json:"parameters"`
+ Strict bool `json:"strict"`
+}
+
+// ModelUsage gives the token counts of a call. Cached is the part of the input
+// tokens that came from the prompt cache.
+type ModelUsage struct {
+ Input int
+ Cached int
+ Output int
+}
+
+// ModelStreamEvent holds the fields of the stream events that the client reads.
+type ModelStreamEvent struct {
+ Type string `json:"type"`
+ Item json.RawMessage `json:"item"`
+ Response *ModelEventResponse `json:"response"`
+ Code string `json:"code"`
+ Message string `json:"message"`
+}
+
+type ModelEventResponse struct {
+ Output []json.RawMessage `json:"output"`
+ Usage *ModelWireUsage `json:"usage"`
+ Error *ModelWireError `json:"error"`
+ IncompleteDetails *struct {
+ Reason string `json:"reason"`
+ } `json:"incomplete_details"`
+}
+
+type ModelWireUsage struct {
+ InputTokens int `json:"input_tokens"`
+ OutputTokens int `json:"output_tokens"`
+ InputTokensDetails struct {
+ CachedTokens int `json:"cached_tokens"`
+ } `json:"input_tokens_details"`
+}
+
+type ModelWireError struct {
+ Code string `json:"code"`
+ Message string `json:"message"`
+}
+
+// newModelClient returns a client for the Responses endpoint at url. It records
+// each call in the usage file at usagePath.
+func newModelClient(url string, auth ModelAuthorizer, cacheKey, usagePath string) *ModelClient {
+ return &ModelClient{
+ url: url,
+ auth: auth,
+ cacheKey: cacheKey,
+ usage: &ModelUsageLog{
+ path: usagePath,
+ },
+ http: &http.Client{},
+ }
+}
+
+// Authorize sets the key as the bearer token.
+func (k ModelAPIKey) Authorize(ctx context.Context, header http.Header) error {
+ header.Set("Authorization", "Bearer "+string(k))
+
+ return ctx.Err()
+}
+
// LimitCalls makes the client refuse each call after the first limit calls.
func (c *ModelClient) LimitCalls(limit int64) {
c.limit = limit
return modelReadStream(httpResponse.Body)
}
-// ModelUsageLog appends one JSON line for each model call to usage.jsonl. Calls
-// can run at once, so a mutex serializes the writes.
-type ModelUsageLog struct {
- mutex sync.Mutex
- path string
-}
-
// Write appends record to the file with one write.
func (u *ModelUsageLog) Write(record ModelUsageRecord) (err error) {
line, err := json.Marshal(record)
return nil
}
-// ModelOutputItem holds the fields of an output item that give its text.
-type ModelOutputItem struct {
- Type string `json:"type"`
- Content []struct {
- Type string `json:"type"`
- Text string `json:"text"`
- } `json:"content"`
-}
-
-// ModelInputPart is a text or image part in a user message or tool result.
-type ModelInputPart struct {
- Type string `json:"type"`
- Text string `json:"text,omitempty"`
- ImageURL string `json:"image_url,omitempty"`
-}
+// modelEndpoint returns the URL and the authorizer of the endpoint in config.
+// The ChatGPT backend uses the token file in dir. The OpenAI API uses the key
+// in OPENAI_API_KEY, and there is no fallback from one to the other. All
+// clients share the authorizer, so one refresh runs at a time.
+func modelEndpoint(config Config, dir string) (string, ModelAuthorizer, error) {
+ if config.Endpoint == configEndpointAPI {
+ key := os.Getenv("OPENAI_API_KEY")
+ if key == "" {
+ return "", nil, errors.New("endpoint api needs OPENAI_API_KEY")
+ }
-// ModelRequest is one model call. Input holds the input items in order, such as
-// modelUserMessage values or output items of an earlier response. Node names
-// the node that a compaction builds, or is nil. Turn names the request of a
-// turn: first or next. Agent names the subagent that makes the call, or is
-// empty for a call of the agent itself. Node, Turn, and Agent go to the usage
-// file, never to the endpoint as fields. Agent also gives the call the cache
-// key of its subagent.
-type ModelRequest struct {
- Model string
- Effort string
- Instructions string
- Tools []ModelTool
- Input []any
- Node *NodeKey
- Turn string
- Agent string
-}
+ return modelAPIURL, ModelAPIKey(key), nil
+ }
-// ModelResponse is the result of a model call. Output holds the output items as
-// the endpoint sent them. Text is the text of the message items.
-type ModelResponse struct {
- Output []json.RawMessage
- Text string
- Usage ModelUsage
+ return modelChatGPTURL, newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer), nil
}
-// ModelRequestBody is the JSON body of a call.
-type ModelRequestBody struct {
- Model string `json:"model"`
- Instructions string `json:"instructions"`
- Input []any `json:"input"`
- Tools []ModelTool `json:"tools"`
- Reasoning ModelReasoning `json:"reasoning"`
- Store bool `json:"store"`
- Stream bool `json:"stream"`
- Include []string `json:"include"`
- PromptCacheKey string `json:"prompt_cache_key"`
-}
+// modelUserMessage returns a user input item with one text part for each text.
+func modelUserMessage(texts ...string) any {
+ content := make([]ModelInputPart, len(texts))
+ for n, text := range texts {
+ content[n] = ModelInputPart{
+ Type: "input_text",
+ Text: text,
+ }
+ }
-type ModelReasoning struct {
- Effort string `json:"effort"`
+ return modelUserParts(content)
}
-// ModelUsageRecord is one line of usage.jsonl. Date is the start of the call,
-// and Seconds its length. Node is the node [l, i] of a compaction. Status is
-// the HTTP status, or 0 when no response came. TextBytes is the size of the
-// text of the reply. Turn names the request of a turn. Agent names the subagent
-// of the call. Quota holds the x-codex-* headers of the response, without the
-// turn state.
-type ModelUsageRecord struct {
- Date string `json:"date"`
- Seconds float64 `json:"seconds"`
- Model string `json:"model"`
- Node *[2]int `json:"node,omitempty"`
- Turn string `json:"turn,omitempty"`
- Agent string `json:"agent,omitempty"`
- Status int `json:"status"`
- InputTokens int `json:"input_tokens"`
- CachedTokens int `json:"cached_tokens"`
- OutputTokens int `json:"output_tokens"`
- TextBytes int `json:"text_bytes"`
- Quota map[string]string `json:"quota,omitempty"`
- Error string `json:"error,omitempty"`
+// modelUserParts keeps text and image parts in one user message.
+func modelUserParts(content []ModelInputPart) any {
+ return struct {
+ Role string `json:"role"`
+ Content []ModelInputPart `json:"content"`
+ }{
+ Role: "user",
+ Content: content,
+ }
}
-// ModelTool is the definition of a function tool.
-type ModelTool struct {
- Type string `json:"type"`
- Name string `json:"name"`
- Description string `json:"description"`
- Parameters any `json:"parameters"`
- Strict bool `json:"strict"`
-}
+// modelOutputText returns the text of the message items of output.
+func modelOutputText(output []json.RawMessage) (string, error) {
+ var text strings.Builder
-// ModelUsage gives the token counts of a call. Cached is the part of the input
-// tokens that came from the prompt cache.
-type ModelUsage struct {
- Input int
- Cached int
- Output int
-}
+ for _, raw := range output {
+ var item ModelOutputItem
-// ModelStreamEvent holds the fields of the stream events that the client reads.
-type ModelStreamEvent struct {
- Type string `json:"type"`
- Item json.RawMessage `json:"item"`
- Response *ModelEventResponse `json:"response"`
- Code string `json:"code"`
- Message string `json:"message"`
-}
+ err := json.Unmarshal(raw, &item)
+ if err != nil {
+ return "", fmt.Errorf("decode output item: %w", err)
+ }
-type ModelEventResponse struct {
- Output []json.RawMessage `json:"output"`
- Usage *ModelWireUsage `json:"usage"`
- Error *ModelWireError `json:"error"`
- IncompleteDetails *struct {
- Reason string `json:"reason"`
- } `json:"incomplete_details"`
-}
+ if item.Type != "message" {
+ continue
+ }
-type ModelWireUsage struct {
- InputTokens int `json:"input_tokens"`
- OutputTokens int `json:"output_tokens"`
- InputTokensDetails struct {
- CachedTokens int `json:"cached_tokens"`
- } `json:"input_tokens_details"`
-}
+ for _, part := range item.Content {
+ if part.Type == "output_text" {
+ text.WriteString(part.Text)
+ }
+ }
+ }
-type ModelWireError struct {
- Code string `json:"code"`
- Message string `json:"message"`
+ return text.String(), nil
}
// modelQuotaHeaders returns the x-codex-* headers, with lowercase names,
reply func(http.ResponseWriter, ModelFakeCall)
}
+func newModelFake(t *testing.T) *ModelFake {
+ t.Helper()
+
+ f := &ModelFake{
+ reply: func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("a line")}, 100, 0, 5))
+ },
+ }
+
+ f.Server = httptest.NewServer(http.HandlerFunc(f.serve))
+ t.Cleanup(f.Close)
+
+ return f
+}
+
// Call returns call number n.
func (f *ModelFake) Call(n int) ModelFakeCall {
f.mutex.Lock()
reply(w, call)
}
-func newModelFake(t *testing.T) *ModelFake {
- t.Helper()
-
- f := &ModelFake{
- reply: func(w http.ResponseWriter, _ ModelFakeCall) {
- modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("a line")}, 100, 0, 5))
- },
- }
-
- f.Server = httptest.NewServer(http.HandlerFunc(f.serve))
- t.Cleanup(f.Close)
-
- return f
-}
-
func modelTestRequest() ModelRequest {
return ModelRequest{
Model: "gpt-test",
probeCallLimit = 660
)
+// ProbeSlotSummarizer lets at most cap(slots) builds call inner at once. A
+// build past the limit waits for a slot.
+type ProbeSlotSummarizer struct {
+ inner Summarizer
+ slots chan struct{}
+}
+
+// Summarize waits for a slot and builds the node with inner.
+func (s ProbeSlotSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
+ select {
+ case s.slots <- struct{}{}:
+ case <-ctx.Done():
+ return "", ctx.Err()
+ }
+
+ defer func() {
+ <-s.slots
+ }()
+
+ return s.inner.Summarize(ctx, task)
+}
+
// probe runs "onechat probe LOG COUNT [BUILDS]". It appends the first COUNT
// messages of LOG, one JSON object {kind, text} per line, to a new chat in
// probe/ in the data directory, and builds its tree with the real compactor.
return nil
}
-// ProbeSlotSummarizer lets at most cap(slots) builds call inner at once. A
-// build past the limit waits for a slot.
-type ProbeSlotSummarizer struct {
- inner Summarizer
- slots chan struct{}
-}
-
-// Summarize waits for a slot and builds the node with inner.
-func (s ProbeSlotSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
- select {
- case s.slots <- struct{}{}:
- case <-ctx.Done():
- return "", ctx.Err()
- }
-
- defer func() {
- <-s.slots
- }()
-
- return s.inner.Summarize(ctx, task)
-}
-
// probeReadLog reads the first count messages of the probe log at path. Source
// dates are ignored because a probe uses its own clock.
func probeReadLog(path string, count int) (entries []DatedMessage, err error) {
// probeMessages is the length of the generated log for the live probe.
const probeMessages = 1000
+var (
+ probePeople = []string{"Mara", "Joel", "Ines", "Tomas", "Priya", "Lena"}
+ probeBreads = []string{"rye", "sourdough", "baguette", "brioche", "spelt", "seed loaf"}
+ probeDays = []string{"Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"}
+ probeThings = []string{"order form", "price list", "delivery route", "stock count", "invoice export", "oven schedule"}
+ probeProblems = []string{"shows the wrong total", "skips the last row", "is slow at noon", "loses the customer note", "rounds the tax down"}
+ probePlaces = []string{"the front page", "the weekly report", "the shared folder", "the admin menu"}
+ probeTasks = []string{"add a field for allergies", "send the invoices on Friday", "split the route in two", "print labels for the shelf"}
+ probeCauses = []string{"the date has no time zone", "the cache keeps old prices", "the loop stops one row early", "the form trims the note"}
+ probeChanges = []string{"the total to use cents", "the route to sort by street", "the export to add the tax line", "the form to keep the note"}
+ probePreferences = []string{"short answers", "a call before a change to prices", "invoices as PDF", "reports in the morning"}
+ probeFiles = []string{"orders/form.go", "prices/list.go", "routes/plan.go", "stock/count.go", "export/invoice.go", "ovens/schedule.go"}
+ probeModules = []string{"orders", "prices", "routes", "stock", "export", "ovens"}
+)
+
// ProbeWriter writes readable made-up messages.
type ProbeWriter struct {
random *rand.Rand
return messages
}
-
-var (
- probePeople = []string{"Mara", "Joel", "Ines", "Tomas", "Priya", "Lena"}
- probeBreads = []string{"rye", "sourdough", "baguette", "brioche", "spelt", "seed loaf"}
- probeDays = []string{"Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"}
- probeThings = []string{"order form", "price list", "delivery route", "stock count", "invoice export", "oven schedule"}
- probeProblems = []string{"shows the wrong total", "skips the last row", "is slow at noon", "loses the customer note", "rounds the tax down"}
- probePlaces = []string{"the front page", "the weekly report", "the shared folder", "the admin menu"}
- probeTasks = []string{"add a field for allergies", "send the invoices on Friday", "split the route in two", "print labels for the shelf"}
- probeCauses = []string{"the date has no time zone", "the cache keeps old prices", "the loop stops one row early", "the form trims the note"}
- probeChanges = []string{"the total to use cents", "the route to sort by street", "the export to add the tax line", "the form to keep the note"}
- probePreferences = []string{"short answers", "a call before a change to prices", "invoices as PDF", "reports in the morning"}
- probeFiles = []string{"orders/form.go", "prices/list.go", "routes/plan.go", "stock/count.go", "export/invoice.go", "ovens/schedule.go"}
- probeModules = []string{"orders", "prices", "routes", "stock", "export", "ovens"}
-)
promptInstructionsHeading = "\n\n# The user's instructions\n\n"
)
-// promptEfforts are the reasoning efforts that spawn can give a subagent.
-var promptEfforts = []string{"low", "medium", "high", "xhigh"}
-
// promptText is the system prompt of the reference, nearly word for word, with
// one added paragraph on the user's instructions file. The reference names its
// own agent, so promptText has the name of the agent and the kind reply in its
whole chat. An [image FILE] line names a stored image. Use zoom(id, 1)
to see its pixels.`
+// promptEfforts are the reasoning efforts that spawn can give a subagent.
+var promptEfforts = []string{"low", "medium", "high", "xhigh"}
+
// prompt returns the system prompt of the reference with one added paragraph on
// the user's instructions file at instructionsPath. It uses the agent name and
// ends with the user's instructions. Without instructions it has neither them
subagentNameDraws = 1000
)
-// newSubagents returns the subagents in dir, one directory for each name. The
-// clock gives the dates of their messages, and draw gives the names of new
-// subagents.
-func newSubagents(dir string, clock func() time.Time, draw func() string) *Subagents {
- return &Subagents{
- dir: dir,
- clock: clock,
- draw: draw,
- open: make(map[string]*Subagent),
- reserved: make(map[string]bool),
- }
-}
-
// Subagents holds the subagents of one agent. Each subagent has a directory
// named for it, which holds its log. A log takes an exclusive lock, so
// Subagents opens the log of a name once and keeps it open until Close. Its
reserved map[string]bool
}
+// Subagent is one subagent, its directory, and its log. Its mutex protects the
+// log, because the run of the subagent writes it while the agent reads it with
+// zoom.
+type Subagent struct {
+ name string
+ dir string
+ mutex sync.Mutex
+ log *Log
+ syncer *Syncer
+}
+
+// newSubagents returns the subagents in dir, one directory for each name. The
+// clock gives the dates of their messages, and draw gives the names of new
+// subagents.
+func newSubagents(dir string, clock func() time.Time, draw func() string) *Subagents {
+ return &Subagents{
+ dir: dir,
+ clock: clock,
+ draw: draw,
+ open: make(map[string]*Subagent),
+ reserved: make(map[string]bool),
+ }
+}
+
// Create makes the directory and the log of a new subagent. It draws a name
// again when the directory of the name exists.
func (s *Subagents) Create(ctx context.Context) (*Subagent, error) {
return err
}
-// Subagent is one subagent, its directory, and its log. Its mutex protects the
-// log, because the run of the subagent writes it while the agent reads it with
-// zoom.
-type Subagent struct {
- name string
- dir string
- mutex sync.Mutex
- log *Log
- syncer *Syncer
-}
-
// Append logs a message of the subagent.
func (a *Subagent) Append(kind, text string) (Message, error) {
a.mutex.Lock()
workers sync.WaitGroup
}
+// SubagentRun is one run of a subagent: the view that its requests get, its
+// effort, the end of its context, the messages of Tell that wait for a request,
+// whether Stop ended it, and the time it started. The mutex of SubagentRuns
+// protects the messages and the stop.
+type SubagentRun struct {
+ subagent *Subagent
+ view []string
+ effort string
+ cancel context.CancelFunc
+ pending []Message
+ stopped bool
+ started time.Time
+}
+
+// SubagentWorkFile is the content of work.json.
+type SubagentWorkFile struct {
+ Running []string `json:"running"`
+}
+
// Spawn starts one subagent for each task and returns their names at once. It
// waits until the view has no pending summary, and gives the view to each
// subagent. A nil effort is the effort of the agent.
return messages
}
-// SubagentRun is one run of a subagent: the view that its requests get, its
-// effort, the end of its context, the messages of Tell that wait for a request,
-// whether Stop ended it, and the time it started. The mutex of SubagentRuns
-// protects the messages and the stop.
-type SubagentRun struct {
- subagent *Subagent
- view []string
- effort string
- cancel context.CancelFunc
- pending []Message
- stopped bool
- started time.Time
-}
-
-// SubagentWorkFile is the content of work.json.
-type SubagentWorkFile struct {
- Running []string `json:"running"`
-}
-
// subagentReport returns the report of the subagent name with its replies: the
// name in brackets, a new line, the replies, and the zoom that gives its whole
// chat.
tasks []Task
}
+func newFakeSummarizer() *FakeSummarizer {
+ return &FakeSummarizer{}
+}
+
func (f *FakeSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
f.mutex.Lock()
return line
}
-
-func newFakeSummarizer() *FakeSummarizer {
- return &FakeSummarizer{}
-}
"golang.org/x/sys/unix"
)
-// openSyncer returns the syncer of the file system that holds dir. It syncs
-// with syncfs, which writes back only that file system.
-func openSyncer(dir string) (*Syncer, error) {
- file, err := os.Open(dir)
- if err != nil {
- return nil, fmt.Errorf("open chat directory: %w", err)
- }
-
- s := newSyncer(func() error {
- return unix.Syncfs(int(file.Fd()))
- })
- s.file = file
-
- return s, nil
-}
-
// Syncer makes the writes of a chat durable by group commit. Each write goes to
// the operating system at once, so a crash of the process loses nothing. After
// each write, Written starts a background sync of the file system of the chat,
workers sync.WaitGroup
}
+// openSyncer returns the syncer of the file system that holds dir. It syncs
+// with syncfs, which writes back only that file system.
+func openSyncer(dir string) (*Syncer, error) {
+ file, err := os.Open(dir)
+ if err != nil {
+ return nil, fmt.Errorf("open chat directory: %w", err)
+ }
+
+ s := newSyncer(func() error {
+ return unix.Syncfs(int(file.Fd()))
+ })
+ s.file = file
+
+ return s, nil
+}
+
+// newSyncer returns a syncer that makes the writes durable with sync.
+func newSyncer(sync func() error) *Syncer {
+ return &Syncer{
+ sync: sync,
+ done: make(chan struct{}),
+ }
+}
+
// Written tells the syncer that a write went to the operating system, and
// starts a sync when none runs.
func (s *Syncer) Written() {
s.done = make(chan struct{})
s.start()
}
-
-// newSyncer returns a syncer that makes the writes durable with sync.
-func newSyncer(sync func() error) *Syncer {
- return &Syncer{
- sync: sync,
- done: make(chan struct{}),
- }
-}
release chan struct{}
}
+func newSyncerGatedSync() *SyncerGatedSync {
+ return &SyncerGatedSync{
+ started: make(chan struct{}, 100),
+ release: make(chan struct{}),
+ }
+}
+
func (g *SyncerGatedSync) Sync() error {
g.mutex.Lock()
return g.count
}
-func newSyncerGatedSync() *SyncerGatedSync {
- return &SyncerGatedSync{
- started: make(chan struct{}, 100),
- release: make(chan struct{}),
- }
-}
-
// syncerRunKilledWriter starts the writer child on the chat in root, kills it
// with SIGKILL after it acknowledged count messages, and returns the number of
// messages that it acknowledged.
telegramTextOnly = "Only text messages are supported."
)
+// TelegramConfig holds the settings of telegram.json: the bot token and the
+// user ID of the owner. The token is a secret, so the file must have mode 0600.
+type TelegramConfig struct {
+ Token string `json:"token"`
+ Owner int64 `json:"owner"`
+}
+
+// TelegramBot connects the private chat of the owner with the agent. It polls
+// the Bot API for new messages and sends owner text, images, and voice notes to
+// the agent. It ignores every other update. Its sender sends the replies, the
+// errors, the notices of the bot, and the user messages of the other clients to
+// the chat, and shows the typing action while a turn runs. Only the replies
+// have a format: their Markdown goes as rich messages, which Telegram renders
+// itself, and as message entities only when Telegram refuses a rich message.
+// The rest is plain text. Tool calls do not go to the chat.
+//
+// The offset file holds the ID of the next update. It moves past an update
+// only after the message of the update is durable in the log, so a restart
+// does not lose a message. A crash between the flush and the offset write
+// makes Telegram send the update again, and the message repeats once.
+//
+// A failed Bot API call waits and tries again, from telegramRetryDelay up to
+// telegramRetryLimit, or as long as the Bot API asks. It never stops the agent.
+// A TelegramOutage logs the first failure and the end of each run of failures,
+// and the retries between them log nothing.
+type TelegramBot struct {
+ api *TelegramAPI
+ voice *VoiceTranscriber
+ owner int64
+ offsetPath string
+ logger *slog.Logger
+
+ mutex sync.Mutex
+ queue []Event
+ ready chan struct{}
+}
+
+// TelegramOutage is one run of failures of a kind of Bot API call. The first
+// failure logs loss, its cause, and the wait of the next try. The retries log
+// nothing. The next answer of Telegram logs back and the length of the outage.
+// loss and back say what the user loses and gets again.
+type TelegramOutage struct {
+ logger *slog.Logger
+ loss string
+ back string
+ start time.Time
+}
+
+// TelegramPart is the text of one message to the chat, at most
+// telegramMessageUnits long, and its entities.
+type TelegramPart struct {
+ text string
+ entities []TelegramEntity
+}
+
+// TelegramOffsetFile is the content of the offset file.
+type TelegramOffsetFile struct {
+ Offset int64 `json:"offset"`
+}
+
// loadTelegramConfig reads telegram.json in dir. It returns false when the file
// does not exist. An error never holds the token.
func loadTelegramConfig(dir string) (TelegramConfig, bool, error) {
}
}
-// TelegramConfig holds the settings of telegram.json: the bot token and the
-// user ID of the owner. The token is a secret, so the file must have mode 0600.
-type TelegramConfig struct {
- Token string `json:"token"`
- Owner int64 `json:"owner"`
-}
-
// validate makes sure that the token has the form of a bot token and that the
// owner is a user ID. The token goes into a URL path, so it can only hold the
// characters of a bot token.
return nil
}
-// TelegramBot connects the private chat of the owner with the agent. It polls
-// the Bot API for new messages and sends owner text, images, and voice notes to
-// the agent. It ignores every other update. Its sender sends the replies, the
-// errors, the notices of the bot, and the user messages of the other clients to
-// the chat, and shows the typing action while a turn runs. Only the replies
-// have a format: their Markdown goes as rich messages, which Telegram renders
-// itself, and as message entities only when Telegram refuses a rich message.
-// The rest is plain text. Tool calls do not go to the chat.
-//
-// The offset file holds the ID of the next update. It moves past an update
-// only after the message of the update is durable in the log, so a restart
-// does not lose a message. A crash between the flush and the offset write
-// makes Telegram send the update again, and the message repeats once.
-//
-// A failed Bot API call waits and tries again, from telegramRetryDelay up to
-// telegramRetryLimit, or as long as the Bot API asks. It never stops the agent.
-// A TelegramOutage logs the first failure and the end of each run of failures,
-// and the retries between them log nothing.
-type TelegramBot struct {
- api *TelegramAPI
- voice *VoiceTranscriber
- owner int64
- offsetPath string
- logger *slog.Logger
-
- mutex sync.Mutex
- queue []Event
- ready chan struct{}
-}
-
// Notify queues event for the chat. It never blocks, so a slow network does not
// stop the agent. The WhatsApp batches and the runs of the schedules do not go
// to the chat: the reply of the agent tells the user what needs the user.
})
}
-// TelegramOutage is one run of failures of a kind of Bot API call. The first
-// failure logs loss, its cause, and the wait of the next try. The retries log
-// nothing. The next answer of Telegram logs back and the length of the outage.
-// loss and back say what the user loses and gets again.
-type TelegramOutage struct {
- logger *slog.Logger
- loss string
- back string
- start time.Time
-}
-
// Fail logs the start of an outage with err and delay, unless an outage runs.
func (o *TelegramOutage) Fail(err error, delay time.Duration) {
if !o.start.IsZero() {
o.start = time.Time{}
}
-// TelegramPart is the text of one message to the chat, at most
-// telegramMessageUnits long, and its entities.
-type TelegramPart struct {
- text string
- entities []TelegramEntity
-}
-
-// TelegramOffsetFile is the content of the offset file.
-type TelegramOffsetFile struct {
- Offset int64 `json:"offset"`
-}
-
// telegramLoadOffset reads the update offset at path. A missing file gives 0,
// which asks for every update that Telegram keeps.
func telegramLoadOffset(path string) (int64, error) {
telegramAnswerBytes = 16 << 20
)
-// newTelegramAPI returns the client of the Bot API at base with token.
-func newTelegramAPI(base, token string) *TelegramAPI {
- return &TelegramAPI{
- base: base,
- token: token,
- http: &http.Client{},
- }
+// TelegramAPI calls the Telegram Bot API. The bot token is a part of each URL,
+// so no error of TelegramAPI holds a URL, a net/http error, or the token.
+type TelegramAPI struct {
+ base string
+ token string
+ http *http.Client
}
-// telegramRetryable tells if a later try of a failed Bot API call can work:
-// after a network error, HTTP 429, or a server error. It returns the wait that
-// the Bot API asks for, or 0.
-func telegramRetryable(err error) (bool, time.Duration) {
- var network *TelegramNetworkError
+// TelegramNetworkError tells that a Bot API call got no valid answer, such as
+// after a network error. A later call can work. Cause tells why in plain words,
+// with no URL and no token.
+type TelegramNetworkError struct {
+ Cause string
+}
- if errors.As(err, &network) {
- return true, 0
- }
+// TelegramError is an answer of the Bot API that tells of a failure. Its text
+// gives the HTTP status and the description, such as "502 Bad Gateway".
+// RetryAfter is the wait that the Bot API asks for after HTTP 429. Refused
+// tells that the answer is a Bot API answer with ok false, not a page of a
+// proxy or of a server error.
+type TelegramError struct {
+ Status int
+ Description string
+ RetryAfter time.Duration
+ Refused bool
+}
- var answer *TelegramError
+// TelegramUpdate is one update of getUpdates. Message is nil for the kinds of
+// updates that are not a new message.
+type TelegramUpdate struct {
+ UpdateID int64 `json:"update_id"`
+ Message *TelegramMessage `json:"message"`
+}
- if errors.As(err, &answer) {
- return answer.Status == http.StatusTooManyRequests || answer.Status >= 500, answer.RetryAfter
- }
+// TelegramEntity is a message entity of sendMessage: a part of the text with a
+// format, such as bold or a link. Offset and Length count UTF-16 code units.
+// URL is the target of a text_link, and Language the language of a pre.
+type TelegramEntity struct {
+ Type string `json:"type"`
+ Offset int `json:"offset"`
+ Length int `json:"length"`
+ URL string `json:"url,omitempty"`
+ Language string `json:"language,omitempty"`
+}
- return false, 0
+// TelegramAnswer is the envelope of each answer of the Bot API.
+type TelegramAnswer struct {
+ OK bool `json:"ok"`
+ Result json.RawMessage `json:"result"`
+ ErrorCode int `json:"error_code"`
+ Description string `json:"description"`
+ Parameters struct {
+ RetryAfter int `json:"retry_after"`
+ } `json:"parameters"`
}
-// telegramRejected tells if the Bot API refused a call with ok false and HTTP
-// 400 or 404. Then the call did nothing: 400 refuses the request, and 404 tells
-// that the method is not there. An error page with no Bot API answer does not
-// tell this, so it is not a refusal.
-func telegramRejected(err error) bool {
- var answer *TelegramError
+// TelegramMessage holds text and supported media metadata of a bot message.
+type TelegramMessage struct {
+ Chat TelegramChat `json:"chat"`
+ From *TelegramUser `json:"from"`
+ Text string `json:"text"`
+ Caption string `json:"caption"`
+ Photo []TelegramPhoto `json:"photo"`
+ Document *TelegramDocument `json:"document"`
+ Voice *TelegramVoice `json:"voice"`
+}
- return errors.As(err, &answer) && answer.Refused && (answer.Status == http.StatusBadRequest || answer.Status == http.StatusNotFound)
+// TelegramVoice identifies a voice note in the authenticated file API.
+type TelegramVoice struct {
+ FileID string `json:"file_id"`
+ FileSize int64 `json:"file_size"`
}
-// TelegramAPI calls the Telegram Bot API. The bot token is a part of each URL,
-// so no error of TelegramAPI holds a URL, a net/http error, or the token.
-type TelegramAPI struct {
- base string
- token string
- http *http.Client
+// TelegramPhoto is one supplied size of a photo.
+type TelegramPhoto struct {
+ FileID string `json:"file_id"`
+ Width int `json:"width"`
+ Height int `json:"height"`
+ FileSize int64 `json:"file_size"`
+}
+
+// TelegramDocument is a file sent without photo compression.
+type TelegramDocument struct {
+ FileID string `json:"file_id"`
+ MimeType string `json:"mime_type"`
+ FileSize int64 `json:"file_size"`
+}
+
+// TelegramChat is the chat of a message. Type is private for a chat with one
+// user.
+type TelegramChat struct {
+ ID int64 `json:"id"`
+ Type string `json:"type"`
+}
+
+// TelegramUser is the sender of a message.
+type TelegramUser struct {
+ ID int64 `json:"id"`
+}
+
+// newTelegramAPI returns the client of the Bot API at base with token.
+func newTelegramAPI(base, token string) *TelegramAPI {
+ return &TelegramAPI{
+ base: base,
+ token: token,
+ http: &http.Client{},
+ }
}
// GetUpdates waits for the updates from offset on, up to
return strings.ReplaceAll(err.Error(), b.token, "TOKEN")
}
-// TelegramNetworkError tells that a Bot API call got no valid answer, such as
-// after a network error. A later call can work. Cause tells why in plain words,
-// with no URL and no token.
-type TelegramNetworkError struct {
- Cause string
-}
-
func (e *TelegramNetworkError) Error() string {
return e.Cause
}
-// TelegramError is an answer of the Bot API that tells of a failure. Its text
-// gives the HTTP status and the description, such as "502 Bad Gateway".
-// RetryAfter is the wait that the Bot API asks for after HTTP 429. Refused
-// tells that the answer is a Bot API answer with ok false, not a page of a
-// proxy or of a server error.
-type TelegramError struct {
- Status int
- Description string
- RetryAfter time.Duration
- Refused bool
-}
-
func (e *TelegramError) Error() string {
return fmt.Sprintf("%d %s", e.Status, e.Description)
}
-// TelegramUpdate is one update of getUpdates. Message is nil for the kinds of
-// updates that are not a new message.
-type TelegramUpdate struct {
- UpdateID int64 `json:"update_id"`
- Message *TelegramMessage `json:"message"`
-}
-
-// TelegramEntity is a message entity of sendMessage: a part of the text with a
-// format, such as bold or a link. Offset and Length count UTF-16 code units.
-// URL is the target of a text_link, and Language the language of a pre.
-type TelegramEntity struct {
- Type string `json:"type"`
- Offset int `json:"offset"`
- Length int `json:"length"`
- URL string `json:"url,omitempty"`
- Language string `json:"language,omitempty"`
-}
-
-// TelegramAnswer is the envelope of each answer of the Bot API.
-type TelegramAnswer struct {
- OK bool `json:"ok"`
- Result json.RawMessage `json:"result"`
- ErrorCode int `json:"error_code"`
- Description string `json:"description"`
- Parameters struct {
- RetryAfter int `json:"retry_after"`
- } `json:"parameters"`
-}
+// telegramRetryable tells if a later try of a failed Bot API call can work:
+// after a network error, HTTP 429, or a server error. It returns the wait that
+// the Bot API asks for, or 0.
+func telegramRetryable(err error) (bool, time.Duration) {
+ var network *TelegramNetworkError
-// TelegramMessage holds text and supported media metadata of a bot message.
-type TelegramMessage struct {
- Chat TelegramChat `json:"chat"`
- From *TelegramUser `json:"from"`
- Text string `json:"text"`
- Caption string `json:"caption"`
- Photo []TelegramPhoto `json:"photo"`
- Document *TelegramDocument `json:"document"`
- Voice *TelegramVoice `json:"voice"`
-}
+ if errors.As(err, &network) {
+ return true, 0
+ }
-// TelegramVoice identifies a voice note in the authenticated file API.
-type TelegramVoice struct {
- FileID string `json:"file_id"`
- FileSize int64 `json:"file_size"`
-}
+ var answer *TelegramError
-// TelegramPhoto is one supplied size of a photo.
-type TelegramPhoto struct {
- FileID string `json:"file_id"`
- Width int `json:"width"`
- Height int `json:"height"`
- FileSize int64 `json:"file_size"`
-}
+ if errors.As(err, &answer) {
+ return answer.Status == http.StatusTooManyRequests || answer.Status >= 500, answer.RetryAfter
+ }
-// TelegramDocument is a file sent without photo compression.
-type TelegramDocument struct {
- FileID string `json:"file_id"`
- MimeType string `json:"mime_type"`
- FileSize int64 `json:"file_size"`
+ return false, 0
}
-// TelegramChat is the chat of a message. Type is private for a chat with one
-// user.
-type TelegramChat struct {
- ID int64 `json:"id"`
- Type string `json:"type"`
-}
+// telegramRejected tells if the Bot API refused a call with ok false and HTTP
+// 400 or 404. Then the call did nothing: 400 refuses the request, and 404 tells
+// that the method is not there. An error page with no Bot API answer does not
+// tell this, so it is not a refusal.
+func telegramRejected(err error) bool {
+ var answer *TelegramError
-// TelegramUser is the sender of a message.
-type TelegramUser struct {
- ID int64 `json:"id"`
+ return errors.As(err, &answer) && answer.Refused && (answer.Status == http.StatusBadRequest || answer.Status == http.StatusNotFound)
}
images map[string][]byte
}
+// newTelegramFake starts a fake Bot API server that stops at the end of the
+// test.
+func newTelegramFake(t *testing.T) *TelegramFake {
+ t.Helper()
+
+ f := &TelegramFake{
+ added: make(chan struct{}, 1),
+ stopped: make(chan struct{}),
+ }
+
+ f.Server = httptest.NewServer(http.HandlerFunc(f.serve))
+ t.Cleanup(f.Close)
+ t.Cleanup(func() {
+ close(f.stopped)
+ })
+
+ return f
+}
+
+// newTelegramLocalFake returns the same fake API without network goroutines.
+// Its RoundTrip handles requests directly for tests with fake time.
+func newTelegramLocalFake(t *testing.T) *TelegramFake {
+ t.Helper()
+
+ f := &TelegramFake{
+ Server: &httptest.Server{URL: "https://example.com"},
+ added: make(chan struct{}, 1),
+ stopped: make(chan struct{}),
+ }
+ t.Cleanup(func() {
+ close(f.stopped)
+ })
+
+ return f
+}
+
// Add makes update available to getUpdates.
func (f *TelegramFake) Add(update TelegramUpdate) {
f.mutex.Lock()
return updates
}
-// newTelegramFake starts a fake Bot API server that stops at the end of the
-// test.
-func newTelegramFake(t *testing.T) *TelegramFake {
- t.Helper()
-
- f := &TelegramFake{
- added: make(chan struct{}, 1),
- stopped: make(chan struct{}),
- }
-
- f.Server = httptest.NewServer(http.HandlerFunc(f.serve))
- t.Cleanup(f.Close)
- t.Cleanup(func() {
- close(f.stopped)
- })
-
- return f
-}
-
-// newTelegramLocalFake returns the same fake API without network goroutines.
-// Its RoundTrip handles requests directly for tests with fake time.
-func newTelegramLocalFake(t *testing.T) *TelegramFake {
- t.Helper()
-
- f := &TelegramFake{
- Server: &httptest.Server{URL: "https://example.com"},
- added: make(chan struct{}, 1),
- stopped: make(chan struct{}),
- }
- t.Cleanup(func() {
- close(f.stopped)
- })
-
- return f
-}
-
// telegramOwnerMessage returns a text message of the owner in the private chat.
func telegramOwnerMessage(text string) *TelegramMessage {
return &TelegramMessage{
telegramTagUnescaper = strings.NewReplacer("&", "&", "<", "<")
)
+// TelegramMarkdownText is the plain text of a Markdown reply and its Telegram
+// entities, while the conversion builds it. Units counts the UTF-16 code units
+// of text, the unit of the entity offsets and lengths.
+type TelegramMarkdownText struct {
+ text strings.Builder
+ units int
+ entities []TelegramEntity
+}
+
+// add appends text.
+func (m *TelegramMarkdownText) add(text string) {
+ m.text.WriteString(text)
+
+ m.units += telegramUTF16Units(text)
+}
+
+// entity keeps entity, unless it is empty, because Telegram refuses an entity
+// of length 0.
+func (m *TelegramMarkdownText) entity(entity TelegramEntity) {
+ if entity.Length > 0 {
+ m.entities = append(m.entities, entity)
+ }
+}
+
+// inline appends one line of Markdown with its inline marks as entities. A
+// backslash before an ASCII punctuation character makes the character plain
+// text.
+func (m *TelegramMarkdownText) inline(line string) {
+ plain := 0
+ index := 0
+ for index < len(line) {
+ if line[index] == '\\' && index+1 < len(line) && strings.IndexByte(markdownPunctuation, line[index+1]) >= 0 {
+ m.add(line[plain:index])
+
+ plain = index + 1
+ index += 2
+
+ continue
+ }
+
+ span, found := markdownMarkSpan(line, index)
+ if !found {
+ index++
+ continue
+ }
+
+ m.add(line[plain:index])
+
+ start := m.units
+ if span.kind == "code" {
+ m.add(span.content)
+ } else {
+ m.inline(span.content)
+ }
+
+ m.entity(TelegramEntity{
+ Type: span.kind,
+ Offset: start,
+ Length: m.units - start,
+ URL: span.url,
+ })
+
+ index = span.end
+ plain = span.end
+ }
+
+ m.add(line[plain:])
+}
+
// telegramMarkdownEntities converts the Markdown of a reply to plain text and
// Telegram message entities. It knows bold (** and __), italic (* and _),
// strikethrough (~~), inline code, fenced code blocks with an optional
return units
}
-// TelegramMarkdownText is the plain text of a Markdown reply and its Telegram
-// entities, while the conversion builds it. Units counts the UTF-16 code units
-// of text, the unit of the entity offsets and lengths.
-type TelegramMarkdownText struct {
- text strings.Builder
- units int
- entities []TelegramEntity
-}
-
-// add appends text.
-func (m *TelegramMarkdownText) add(text string) {
- m.text.WriteString(text)
-
- m.units += telegramUTF16Units(text)
-}
-
-// entity keeps entity, unless it is empty, because Telegram refuses an entity
-// of length 0.
-func (m *TelegramMarkdownText) entity(entity TelegramEntity) {
- if entity.Length > 0 {
- m.entities = append(m.entities, entity)
- }
-}
-
-// inline appends one line of Markdown with its inline marks as entities. A
-// backslash before an ASCII punctuation character makes the character plain
-// text.
-func (m *TelegramMarkdownText) inline(line string) {
- plain := 0
- index := 0
- for index < len(line) {
- if line[index] == '\\' && index+1 < len(line) && strings.IndexByte(markdownPunctuation, line[index+1]) >= 0 {
- m.add(line[plain:index])
-
- plain = index + 1
- index += 2
-
- continue
- }
-
- span, found := markdownMarkSpan(line, index)
- if !found {
- index++
- continue
- }
-
- m.add(line[plain:index])
-
- start := m.units
- if span.kind == "code" {
- m.add(span.content)
- } else {
- m.inline(span.content)
- }
-
- m.entity(TelegramEntity{
- Type: span.kind,
- Offset: start,
- Length: m.units - start,
- URL: span.url,
- })
-
- index = span.end
- plain = span.end
- }
-
- m.add(line[plain:])
-}
-
// telegramRewriteProse returns markdown with rewrite applied to each part of a
// line outside inline code and fenced code blocks. A code block that does not
// close runs to the end, as in Markdown.
imageShow *ImageShow
}
+// ToolZoomArguments are the arguments of zoom. A Name that is not nil names the
+// child whose view or the subagent whose whole log zoom gives. ID and N are
+// then not used.
+type ToolZoomArguments struct {
+ ID int `json:"id"`
+ N int `json:"n"`
+ Name *string `json:"name"`
+}
+
+// ToolDateArguments are the arguments of date.
+type ToolDateArguments struct {
+ ID int `json:"id"`
+}
+
+// ToolShellArguments are the arguments of shell.
+type ToolShellArguments struct {
+ Command string `json:"command"`
+}
+
+// ToolPathArguments are the arguments of read.
+type ToolPathArguments struct {
+ Path string `json:"path"`
+}
+
+// ToolWriteArguments are the arguments of write.
+type ToolWriteArguments struct {
+ Path string `json:"path"`
+ Text string `json:"text"`
+}
+
+// ToolEditArguments are the arguments of edit.
+type ToolEditArguments struct {
+ Path string `json:"path"`
+ Old string `json:"old"`
+ New string `json:"new"`
+}
+
+// ToolSpawnArguments are the arguments of spawn. A nil Effort is the effort of
+// the agent.
+type ToolSpawnArguments struct {
+ Tasks []string `json:"tasks"`
+ Effort *string `json:"effort"`
+}
+
+// ToolAgentCreateArguments are the arguments of agent_create.
+type ToolAgentCreateArguments struct {
+ Name string `json:"name"`
+ Responsibility string `json:"responsibility"`
+}
+
+// ToolRouteArguments are the arguments of route.
+type ToolRouteArguments struct {
+ Source string `json:"source"`
+ Agent string `json:"agent"`
+}
+
+// ToolRouteSourceArguments are the arguments of unroute.
+type ToolRouteSourceArguments struct {
+ Source string `json:"source"`
+}
+
+// ToolTellArguments are the arguments of tell.
+type ToolTellArguments struct {
+ Name string `json:"name"`
+ Message string `json:"message"`
+}
+
+// ToolImageShowArguments are the arguments of image_show.
+type ToolImageShowArguments struct {
+ Image string `json:"image"`
+ Caption string `json:"caption"`
+}
+
+// ToolWhatsAppFindArguments are the arguments of whatsapp_find.
+type ToolWhatsAppFindArguments struct {
+ Name string `json:"name"`
+}
+
+// ToolWhatsAppSendArguments are the arguments of whatsapp_send.
+type ToolWhatsAppSendArguments struct {
+ ID string `json:"id"`
+ Chat string `json:"chat"`
+ Text string `json:"text"`
+ Image string `json:"image"`
+}
+
+// ToolScheduleSetArguments are the arguments of schedule_set. Exactly one of At
+// and Cron is not nil.
+type ToolScheduleSetArguments struct {
+ Text string `json:"text"`
+ At *string `json:"at"`
+ Cron *string `json:"cron"`
+}
+
+// ToolScheduleRemoveArguments are the arguments of schedule_remove.
+type ToolScheduleRemoveArguments struct {
+ ID int `json:"id"`
+}
+
+// ToolHeadTail keeps the first and the last toolClipBytes bytes written to it,
+// and counts the bytes between them that it drops.
+type ToolHeadTail struct {
+ head []byte
+ tail []byte
+ dropped int
+}
+
// Call logs the call of the tool name with its JSON arguments, makes the log
// durable, runs the tool, and logs its clipped text. It returns that text or
// the text and image parts of a single-message zoom. Pixels never enter the
return filepath.Join(b.workdir, path)
}
-// ToolZoomArguments are the arguments of zoom. A Name that is not nil names the
-// child whose view or the subagent whose whole log zoom gives. ID and N are
-// then not used.
-type ToolZoomArguments struct {
- ID int `json:"id"`
- N int `json:"n"`
- Name *string `json:"name"`
-}
-
-// ToolDateArguments are the arguments of date.
-type ToolDateArguments struct {
- ID int `json:"id"`
-}
-
-// ToolShellArguments are the arguments of shell.
-type ToolShellArguments struct {
- Command string `json:"command"`
-}
-
-// ToolPathArguments are the arguments of read.
-type ToolPathArguments struct {
- Path string `json:"path"`
-}
-
-// ToolWriteArguments are the arguments of write.
-type ToolWriteArguments struct {
- Path string `json:"path"`
- Text string `json:"text"`
-}
-
-// ToolEditArguments are the arguments of edit.
-type ToolEditArguments struct {
- Path string `json:"path"`
- Old string `json:"old"`
- New string `json:"new"`
-}
-
-// ToolSpawnArguments are the arguments of spawn. A nil Effort is the effort of
-// the agent.
-type ToolSpawnArguments struct {
- Tasks []string `json:"tasks"`
- Effort *string `json:"effort"`
-}
-
-// ToolAgentCreateArguments are the arguments of agent_create.
-type ToolAgentCreateArguments struct {
- Name string `json:"name"`
- Responsibility string `json:"responsibility"`
-}
-
-// ToolRouteArguments are the arguments of route.
-type ToolRouteArguments struct {
- Source string `json:"source"`
- Agent string `json:"agent"`
-}
-
-// ToolRouteSourceArguments are the arguments of unroute.
-type ToolRouteSourceArguments struct {
- Source string `json:"source"`
-}
-
-// ToolTellArguments are the arguments of tell.
-type ToolTellArguments struct {
- Name string `json:"name"`
- Message string `json:"message"`
-}
-
-// ToolImageShowArguments are the arguments of image_show.
-type ToolImageShowArguments struct {
- Image string `json:"image"`
- Caption string `json:"caption"`
-}
-
-// ToolWhatsAppFindArguments are the arguments of whatsapp_find.
-type ToolWhatsAppFindArguments struct {
- Name string `json:"name"`
-}
-
-// ToolWhatsAppSendArguments are the arguments of whatsapp_send.
-type ToolWhatsAppSendArguments struct {
- ID string `json:"id"`
- Chat string `json:"chat"`
- Text string `json:"text"`
- Image string `json:"image"`
-}
-
-// ToolScheduleSetArguments are the arguments of schedule_set. Exactly one of At
-// and Cron is not nil.
-type ToolScheduleSetArguments struct {
- Text string `json:"text"`
- At *string `json:"at"`
- Cron *string `json:"cron"`
-}
-
-// ToolScheduleRemoveArguments are the arguments of schedule_remove.
-type ToolScheduleRemoveArguments struct {
- ID int `json:"id"`
-}
-
-// ToolHeadTail keeps the first and the last toolClipBytes bytes written to it,
-// and counts the bytes between them that it drops.
-type ToolHeadTail struct {
- head []byte
- tail []byte
- dropped int
-}
-
// String returns the output, clipped when it is too long. Invalid UTF-8 becomes
// the replacement character.
func (h *ToolHeadTail) String() string {
events []string
}
+// newTestToolbox returns a toolbox with log, an empty chat, and the working
+// directory dir.
+func newTestToolbox(t *testing.T, log ToolLog, dir string) *Toolbox {
+ t.Helper()
+
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog())
+ require.NoError(t, err)
+
+ t.Cleanup(func() {
+ require.NoError(t, memory.Close())
+ })
+
+ return &Toolbox{
+ log: log,
+ chat: memory,
+ subagents: newTestSubagents(t),
+ workdir: dir,
+ }
+}
+
func (r *ToolRecordingLog) Append(kind, text string) (Message, error) {
r.mutex.Lock()
defer r.mutex.Unlock()
return append([]string(nil), r.events...)
}
-
-// newTestToolbox returns a toolbox with log, an empty chat, and the working
-// directory dir.
-func newTestToolbox(t *testing.T, log ToolLog, dir string) *Toolbox {
- t.Helper()
-
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog())
- require.NoError(t, err)
-
- t.Cleanup(func() {
- require.NoError(t, memory.Close())
- })
-
- return &Toolbox{
- log: log,
- chat: memory,
- subagents: newTestSubagents(t),
- workdir: dir,
- }
-}
// the node line, the node index, and the node level.
const treeRecordBytes = 32
+// Tree holds the built nodes of one chat. It writes one JSON line for each node
+// to tree/YYYY-MM-DD.jsonl, named for the local day of the build. Each node is
+// built once and never again. A node of a higher level is built only after both
+// of its children. After a failed write, the tree refuses all writes until it
+// is opened again. The caller holds the lock of the chat and serializes all
+// calls.
+//
+// The tree keeps no node in memory. The index index/tree.idx lists the node
+// lines in file order, and index/tree-L.idx holds the position of node (L, i)
+// in record i, with a zero record for a node that is not built. At open the
+// tree indexes the lines after the last indexed one, so a missing index is
+// built again from the day files.
+type Tree struct {
+ clock func() time.Time
+ journal Journal
+ dir string
+ order *Records
+ levels map[int]*Records
+ failure error
+}
+
+// Node is one line of the summary tree. Node (l, i) covers the 2^l messages
+// from i*2^l on. A node of level 0 holds one message. A node of a higher level
+// merges its two children, (l-1, 2i) and (l-1, 2i+1). Size is the UTF-8 byte
+// length of Text.
+type Node struct {
+ Level int `json:"l"`
+ Index int `json:"i"`
+ Text string `json:"text"`
+ Size int `json:"size"`
+}
+
+// NodeKey names the node (l, i).
+type NodeKey struct {
+ level int
+ index int
+}
+
// openTree opens the tree of the chat in root and indexes its new lines. A chat
// of total messages cannot have a node that covers a later message.
func openTree(ctx context.Context, root string, total int, clock func() time.Time) (*Tree, error) {
return t, nil
}
-// leafSource returns the text that the leaf of message holds or summarizes.
-func leafSource(message Message) string {
- return message.Kind + ": " + message.Text
-}
-
-// freeLeaf returns a text-only leaf when its source fits in a node. Images need
-// a model description even when their text fits.
-func freeLeaf(message Message) (Node, bool) {
- text := leafSource(message)
- if len(text) > nodeBytes || len(imageFiles(message.Text)) > 0 {
- return Node{}, false
- }
-
- return Node{
- Level: 0,
- Index: message.ID,
- Text: text,
- Size: len(text),
- }, true
-}
-
-// freeMerge returns the parent of two sibling nodes when both texts, joined by
-// a newline, fit in a node. Such a parent needs no model call.
-func freeMerge(left, right Node) (Node, bool) {
- if left.Size+1+right.Size > nodeBytes {
- return Node{}, false
- }
-
- text := left.Text + "\n" + right.Text
-
- return Node{
- Level: left.Level + 1,
- Index: left.Index / 2,
- Text: text,
- Size: len(text),
- }, true
-}
-
-// nodeEnd returns the ID after the last message of node (l, i).
-func nodeEnd(level, index int) int {
- return (index + 1) << level
-}
-
-// Tree holds the built nodes of one chat. It writes one JSON line for each node
-// to tree/YYYY-MM-DD.jsonl, named for the local day of the build. Each node is
-// built once and never again. A node of a higher level is built only after both
-// of its children. After a failed write, the tree refuses all writes until it
-// is opened again. The caller holds the lock of the chat and serializes all
-// calls.
-//
-// The tree keeps no node in memory. The index index/tree.idx lists the node
-// lines in file order, and index/tree-L.idx holds the position of node (L, i)
-// in record i, with a zero record for a node that is not built. At open the
-// tree indexes the lines after the last indexed one, so a missing index is
-// built again from the day files.
-type Tree struct {
- clock func() time.Time
- journal Journal
- dir string
- order *Records
- levels map[int]*Records
- failure error
-}
-
// Close closes the files of the tree.
func (t *Tree) Close() error {
err := t.journal.Close()
return t.order.count
}
-// Node is one line of the summary tree. Node (l, i) covers the 2^l messages
-// from i*2^l on. A node of level 0 holds one message. A node of a higher level
-// merges its two children, (l-1, 2i) and (l-1, 2i+1). Size is the UTF-8 byte
-// length of Text.
-type Node struct {
- Level int `json:"l"`
- Index int `json:"i"`
- Text string `json:"text"`
- Size int `json:"size"`
+// leafSource returns the text that the leaf of message holds or summarizes.
+func leafSource(message Message) string {
+ return message.Kind + ": " + message.Text
}
-// NodeKey names the node (l, i).
-type NodeKey struct {
- level int
- index int
+// freeLeaf returns a text-only leaf when its source fits in a node. Images need
+// a model description even when their text fits.
+func freeLeaf(message Message) (Node, bool) {
+ text := leafSource(message)
+ if len(text) > nodeBytes || len(imageFiles(message.Text)) > 0 {
+ return Node{}, false
+ }
+
+ return Node{
+ Level: 0,
+ Index: message.ID,
+ Text: text,
+ Size: len(text),
+ }, true
+}
+
+// freeMerge returns the parent of two sibling nodes when both texts, joined by
+// a newline, fit in a node. Such a parent needs no model call.
+func freeMerge(left, right Node) (Node, bool) {
+ if left.Size+1+right.Size > nodeBytes {
+ return Node{}, false
+ }
+
+ text := left.Text + "\n" + right.Text
+
+ return Node{
+ Level: left.Level + 1,
+ Index: left.Index / 2,
+ Text: text,
+ Size: len(text),
+ }, true
+}
+
+// nodeEnd returns the ID after the last message of node (l, i).
+func nodeEnd(level, index int) int {
+ return (index + 1) << level
}
// treeDecodeRecord reads the key and the position of a record of
"time"
)
+// Event tells the clients what the agent does as it happens. Kind is the kind
+// of a logged message (user, reply, tool, echo, note, work), or one of ready,
+// busy, idle, error, and subagent, which tells that the subagent Text starts a
+// run. ID is the ID of a logged message. Source names the client that sent a
+// user message, such as terminal or telegram, so that each client can show the
+// messages of the others, or the subagent of a report. Status tells how the run
+// of a report ended: finished, stopped, or failed.
+type Event struct {
+ Kind string `json:"kind"`
+ ID int `json:"id,omitempty"`
+ Text string `json:"text,omitempty"`
+ Source string `json:"source,omitempty"`
+ Status string `json:"status,omitempty"`
+}
+
+// TurnLoop runs the turns of one agent. A user message goes to the log when it
+// comes, and it wakes the loop. A turn waits until every message before it is
+// summarized. Then the turn runs a Chain whose first request holds the prompt,
+// the tools, and the input of turnInput.
+//
+// The view goes out as the messages of a ViewLayout, so the first request of a
+// turn reads the view of the turn before from the prompt cache. User messages
+// that come during a turn join the next request of its chain after the tool
+// results, and so do the reports of the subagents. The subagents run until the
+// loop ends. Parent and child messages stay queued until the turn finishes.
+// active belongs to the running turn. The mutex protects pending, cancel, and
+// failure.
+type TurnLoop struct {
+ agent *Agent
+ memory *Memory
+ client ChainClient
+ tools *Toolbox
+ runs *SubagentRuns
+ layout *ViewLayout
+ clock func() time.Time
+ logger *slog.Logger
+ notify func(Event)
+ wake chan struct{}
+ messages *DaemonMessages
+ openChild func(context.Context, *Agent) (*DaemonAgent, error)
+ active []Message
+
+ mutex sync.Mutex
+ pending []Message
+ cancel context.CancelFunc
+ failure error
+}
+
// newTurnLoop returns the loop of agent with its memory, its subagents, and the
// WhatsApp link and the schedules of its tools. Its calls go through client,
// and notify gets each event. The clock gives the date of a turn. logger gets
return l, nil
}
-// Event tells the clients what the agent does as it happens. Kind is the kind
-// of a logged message (user, reply, tool, echo, note, work), or one of ready,
-// busy, idle, error, and subagent, which tells that the subagent Text starts a
-// run. ID is the ID of a logged message. Source names the client that sent a
-// user message, such as terminal or telegram, so that each client can show the
-// messages of the others, or the subagent of a report. Status tells how the run
-// of a report ended: finished, stopped, or failed.
-type Event struct {
- Kind string `json:"kind"`
- ID int `json:"id,omitempty"`
- Text string `json:"text,omitempty"`
- Source string `json:"source,omitempty"`
- Status string `json:"status,omitempty"`
-}
-
-// TurnLoop runs the turns of one agent. A user message goes to the log when it
-// comes, and it wakes the loop. A turn waits until every message before it is
-// summarized. Then the turn runs a Chain whose first request holds the prompt,
-// the tools, and the input of turnInput.
-//
-// The view goes out as the messages of a ViewLayout, so the first request of a
-// turn reads the view of the turn before from the prompt cache. User messages
-// that come during a turn join the next request of its chain after the tool
-// results, and so do the reports of the subagents. The subagents run until the
-// loop ends. Parent and child messages stay queued until the turn finishes.
-// active belongs to the running turn. The mutex protects pending, cancel, and
-// failure.
-type TurnLoop struct {
- agent *Agent
- memory *Memory
- client ChainClient
- tools *Toolbox
- runs *SubagentRuns
- layout *ViewLayout
- clock func() time.Time
- logger *slog.Logger
- notify func(Event)
- wake chan struct{}
- messages *DaemonMessages
- openChild func(context.Context, *Agent) (*DaemonAgent, error)
- active []Message
-
- mutex sync.Mutex
- pending []Message
- cancel context.CancelFunc
- failure error
-}
-
// Submit logs a user message from the client source and wakes the loop. The
// message joins the next request: the first one of a new turn, or the next one
// of a running turn.
usagePath string
}
+// TurnEventRecorder keeps the events of a loop.
+type TurnEventRecorder struct {
+ mutex sync.Mutex
+ events []Event
+}
+
+func newTestTurnLoop(t *testing.T, server *ModelFake) *TurnTestLoop {
+ t.Helper()
+
+ return newTestTurnLoopWith(t, server, newFakeSummarizer())
+}
+
+// newTestTurnLoopWith returns a running loop on a new agent whose builds use
+// summarizer and whose commands run in a temporary directory.
+func newTestTurnLoopWith(t *testing.T, server *ModelFake, summarizer Summarizer) *TurnTestLoop {
+ t.Helper()
+
+ agent, err := openAgent(filepath.Join(t.TempDir(), "agent"))
+ require.NoError(t, err)
+
+ agent.config.Workdir = t.TempDir()
+ clock := newTestClock()
+ memory, err := openMemory(t.Context(), t.TempDir(), clock.Now, summarizer, discardLog())
+ require.NoError(t, err)
+
+ test := &TurnTestLoop{
+ agent: agent,
+ memory: memory,
+ events: &TurnEventRecorder{},
+ logs: &SyncBuffer{},
+ usagePath: filepath.Join(t.TempDir(), "usage.jsonl"),
+ }
+
+ client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), test.usagePath)
+ whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), clock)
+ test.loop, err = newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, clock.Now), client, clock.Now, daemonTestLogger(test.logs), test.events.Add)
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithCancel(t.Context())
+ done := make(chan error)
+ go func() {
+ done <- test.loop.Run(ctx)
+ }()
+
+ t.Cleanup(func() {
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ require.NoError(t, memory.Close())
+ })
+
+ return test
+}
+
// Log returns the logged messages as "kind: text".
func (l *TurnTestLoop) Log(t *testing.T) []string {
t.Helper()
return lines
}
-// TurnEventRecorder keeps the events of a loop.
-type TurnEventRecorder struct {
- mutex sync.Mutex
- events []Event
-}
-
// WaitFor waits until count events of kind came.
func (r *TurnEventRecorder) WaitFor(t *testing.T, kind string, count int) {
t.Helper()
r.events = append(r.events, event)
}
-func newTestTurnLoop(t *testing.T, server *ModelFake) *TurnTestLoop {
- t.Helper()
-
- return newTestTurnLoopWith(t, server, newFakeSummarizer())
-}
-
// turnInputTexts returns the text of each user message of the input of call.
func turnInputTexts(t *testing.T, call ModelFakeCall) []string {
t.Helper()
return texts
}
-// newTestTurnLoopWith returns a running loop on a new agent whose builds use
-// summarizer and whose commands run in a temporary directory.
-func newTestTurnLoopWith(t *testing.T, server *ModelFake, summarizer Summarizer) *TurnTestLoop {
- t.Helper()
-
- agent, err := openAgent(filepath.Join(t.TempDir(), "agent"))
- require.NoError(t, err)
-
- agent.config.Workdir = t.TempDir()
- clock := newTestClock()
- memory, err := openMemory(t.Context(), t.TempDir(), clock.Now, summarizer, discardLog())
- require.NoError(t, err)
-
- test := &TurnTestLoop{
- agent: agent,
- memory: memory,
- events: &TurnEventRecorder{},
- logs: &SyncBuffer{},
- usagePath: filepath.Join(t.TempDir(), "usage.jsonl"),
- }
-
- client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), test.usagePath)
- whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), clock)
- test.loop, err = newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, clock.Now), client, clock.Now, daemonTestLogger(test.logs), test.events.Add)
- require.NoError(t, err)
-
- ctx, cancel := context.WithCancel(t.Context())
- done := make(chan error)
- go func() {
- done <- test.loop.Run(ctx)
- }()
-
- t.Cleanup(func() {
- cancel()
- require.ErrorIs(t, <-done, context.Canceled)
- require.NoError(t, memory.Close())
- })
-
- return test
-}
-
func modelReasoningItem(id, encrypted string) map[string]any {
return map[string]any{
"type": "reasoning",
viewPending = "(not summarized yet: zoom it)"
)
-// newView returns an empty view with the limits high and low.
-func newView(high, low int) *View {
- return &View{
- high: high,
- low: low,
- nodes: make(map[NodeKey]Node),
- }
-}
-
-// restoreView returns the view of state with the limits high and low. Each line
-// must continue the one before it, and each line above level 0 must be built.
-func restoreView(state ViewState, high, low int, tree ViewNodeReader) (*View, error) {
- v := newView(high, low)
- v.merging = state.Merging
- for _, pair := range state.Lines {
- line := NodeKey{pair[0], pair[1]}
- if line.level < 0 || line.level >= bits.UintSize-2 || line.index < 0 || line.index<<line.level != v.end {
- return nil, fmt.Errorf("view line (%d, %d) does not start at message %d", line.level, line.index, v.end)
- }
-
- built, err := v.load(tree, line)
- if err != nil {
- return nil, err
- }
-
- if line.level > 0 && !built {
- return nil, fmt.Errorf("view line (%d, %d) is not built", line.level, line.index)
- }
-
- v.lines = append(v.lines, line)
- v.end = nodeEnd(line.level, line.index)
- }
-
- return v, nil
-}
-
// View is a list of tree nodes that covers messages 0 to end-1 without a gap,
// oldest first. Each new message appends its leaf as a line. Once the size of
// the view goes above high, one batch merges the most due sibling pairs until
merging bool
}
+// ViewState is the saved form of a View. The limits are not part of it.
+type ViewState struct {
+ Lines [][2]int `json:"lines"`
+ Merging bool `json:"merging"`
+}
+
+// ViewNodeReader gives the built nodes of a tree.
+type ViewNodeReader interface {
+ Built(level, index int) (bool, error)
+ Node(level, index int) (Node, bool, error)
+}
+
+// newView returns an empty view with the limits high and low.
+func newView(high, low int) *View {
+ return &View{
+ high: high,
+ low: low,
+ nodes: make(map[NodeKey]Node),
+ }
+}
+
// State returns the saved form of the view.
func (v *View) State() ViewState {
state := ViewState{
return node.Size
}
-// ViewState is the saved form of a View. The limits are not part of it.
-type ViewState struct {
- Lines [][2]int `json:"lines"`
- Merging bool `json:"merging"`
-}
+// restoreView returns the view of state with the limits high and low. Each line
+// must continue the one before it, and each line above level 0 must be built.
+func restoreView(state ViewState, high, low int, tree ViewNodeReader) (*View, error) {
+ v := newView(high, low)
+ v.merging = state.Merging
+ for _, pair := range state.Lines {
+ line := NodeKey{pair[0], pair[1]}
+ if line.level < 0 || line.level >= bits.UintSize-2 || line.index < 0 || line.index<<line.level != v.end {
+ return nil, fmt.Errorf("view line (%d, %d) does not start at message %d", line.level, line.index, v.end)
+ }
-// ViewNodeReader gives the built nodes of a tree.
-type ViewNodeReader interface {
- Built(level, index int) (bool, error)
- Node(level, index int) (Node, bool, error)
+ built, err := v.load(tree, line)
+ if err != nil {
+ return nil, err
+ }
+
+ if line.level > 0 && !built {
+ return nil, fmt.Errorf("view line (%d, %d) is not built", line.level, line.index)
+ }
+
+ v.lines = append(v.lines, line)
+ v.end = nodeEnd(line.level, line.index)
+ }
+
+ return v, nil
}
// agent.
const viewLayoutFileName = "layout.json"
+// ViewLayout splits the view of a turn request into user messages that only
+// grow between batches. The view goes out as the same messages as in the
+// request before, while their lines give the same text, and the new lines
+// follow as one new message. So the end of the view of each earlier turn stays
+// a message end. On the ChatGPT backend, the first request of a turn with this
+// layout read the view of the turn before from the prompt cache in each of 10
+// turns, and with the view in one message it missed once in 5.
+//
+// The first message starts with "<chat>\n". The closing tag goes into the
+// state message after the view, so that no view message changes when lines
+// come. The layout keeps the line count and a hash of the text of each message
+// in layout.json, so it outlives a restart.
+type ViewLayout struct {
+ path string
+ Messages []ViewLayoutMessage `json:"messages"`
+}
+
+// ViewLayoutMessage is one view message: its number of lines and the hash of
+// its text.
+type ViewLayoutMessage struct {
+ Lines int `json:"lines"`
+ Hash string `json:"hash"`
+}
+
// loadViewLayout reads the layout at path. A missing file gives an empty
// layout.
func loadViewLayout(path string) (*ViewLayout, error) {
return layout, nil
}
-// ViewLayout splits the view of a turn request into user messages that only
-// grow between batches. The view goes out as the same messages as in the
-// request before, while their lines give the same text, and the new lines
-// follow as one new message. So the end of the view of each earlier turn stays
-// a message end. On the ChatGPT backend, the first request of a turn with this
-// layout read the view of the turn before from the prompt cache in each of 10
-// turns, and with the view in one message it missed once in 5.
-//
-// The first message starts with "<chat>\n". The closing tag goes into the
-// state message after the view, so that no view message changes when lines
-// come. The layout keeps the line count and a hash of the text of each message
-// in layout.json, so it outlives a restart.
-type ViewLayout struct {
- path string
- Messages []ViewLayoutMessage `json:"messages"`
-}
-
// Split returns the view messages for lines and saves the layout. It keeps the
// earlier messages up to the first one whose text differs, and puts the rest
// of the lines into one new message.
return texts, nil
}
-// ViewLayoutMessage is one view message: its number of lines and the hash of
-// its text.
-type ViewLayoutMessage struct {
- Lines int `json:"lines"`
- Hash string `json:"hash"`
-}
-
// viewMessage returns the text of a view message with lines. The first one
// opens the chat.
func viewMessage(first bool, lines []string) string {
// ViewBuiltTree is a tree with each node built, with 1 byte of text.
type ViewBuiltTree struct{}
-func (v ViewBuiltTree) Built(_, _ int) (bool, error) {
- return true, nil
-}
-
-func (ViewBuiltTree) Node(level, index int) (Node, bool, error) {
- return treeTestNode(level, index, "a"), true, nil
-}
-
// ViewPushEntry is one entry of the rollback state list of the reference
// design, newest first. Under push alone, life stays 0, so the entry leaves it
// out.
older *ViewPushEntry
}
+// ViewPartTree is a tree with the built nodes of its sizes.
+type ViewPartTree struct {
+ sizes map[NodeKey]int
+}
+
+func (v ViewBuiltTree) Built(_, _ int) (bool, error) {
+ return true, nil
+}
+
+func (ViewBuiltTree) Node(level, index int) (Node, bool, error) {
+ return treeTestNode(level, index, "a"), true, nil
+}
+
// starts returns the states of the list, oldest first.
func (e *ViewPushEntry) starts() []int {
var starts []int
return starts
}
-// ViewPartTree is a tree with the built nodes of its sizes.
-type ViewPartTree struct {
- sizes map[NodeKey]int
-}
-
func (p ViewPartTree) Built(level, index int) (bool, error) {
_, built := p.sizes[NodeKey{level, index}]
return built, nil
SendMessage(ctx context.Context, to types.JID, message *waE2E.Message, extra ...whatsmeow.SendRequestExtra) (whatsmeow.SendResponse, error)
}
-// openWhatsApp opens the WhatsApp session in the data directory dir. Without
-// the session file, WhatsApp is not linked. The session gives full access to
-// the account, so a file that another user can read is an error. The clock
-// gives the times of the status, logger gets the failures, and newClient makes
-// the client of a device with its event handler.
-func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logger *slog.Logger, newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient) (*WhatsApp, error) {
- w := &WhatsApp{
- ctx: ctx,
- dir: filepath.Join(dir, whatsappDirName),
- clock: clock,
- logger: logger,
- newClient: newClient,
- requests: make(chan WhatsAppRequest, whatsappRequests),
- logouts: make(chan struct{}, 1),
- groups: newWhatsAppGroups(logger),
- }
-
- var err error
-
- w.state, err = openWhatsAppState(filepath.Join(w.dir, whatsappStateFileName))
- if err != nil {
- return nil, err
- }
-
- w.inbox = newWhatsAppInbox(w.state, clock, w.groups, &ImageStore{dir: filepath.Join(dir, imageDirName)}, logger)
-
- info, err := os.Stat(w.sessionPath())
- if errors.Is(err, os.ErrNotExist) {
- return w, nil
- }
-
- if err != nil {
- return nil, fmt.Errorf("read WhatsApp session: %w", err)
- }
-
- if info.Mode().Perm()&0o077 != 0 {
- return nil, fmt.Errorf("%s holds the WhatsApp session, so its mode must be 0600", filepath.Join(whatsappDirName, whatsappSessionFileName))
- }
-
- err = w.openSession(ctx)
- if err != nil {
- return nil, err
- }
-
- if w.device.ID != nil {
- w.linked = true
- w.inbox.waitForCatchup()
-
- w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
- }
-
- return w, nil
-}
-
// WhatsApp is the link of the daemon to the WhatsApp account of the user, as a
// linked device. The session database of the library in dir holds the keys of
// the device. Run keeps a linked device connected, and the library connects
closed bool
}
+// WhatsAppLink is a copy of the link of the device at one time, for the parts
+// of WhatsApp that do not own it: the groups, the batches, find, and send.
+type WhatsAppLink struct {
+ linked bool
+ device *store.Device
+ client WhatsAppClient
+ connected bool
+ own []string
+}
+
+// WhatsAppRequest is a pair or unlink command for Run, with the function that
+// gives its answers to the client that asked. A pairing has the phone number.
+type WhatsAppRequest struct {
+ phone string
+ unlink bool
+ answer func(text string)
+}
+
+// openWhatsApp opens the WhatsApp session in the data directory dir. Without
+// the session file, WhatsApp is not linked. The session gives full access to
+// the account, so a file that another user can read is an error. The clock
+// gives the times of the status, logger gets the failures, and newClient makes
+// the client of a device with its event handler.
+func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logger *slog.Logger, newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient) (*WhatsApp, error) {
+ w := &WhatsApp{
+ ctx: ctx,
+ dir: filepath.Join(dir, whatsappDirName),
+ clock: clock,
+ logger: logger,
+ newClient: newClient,
+ requests: make(chan WhatsAppRequest, whatsappRequests),
+ logouts: make(chan struct{}, 1),
+ groups: newWhatsAppGroups(logger),
+ }
+
+ var err error
+
+ w.state, err = openWhatsAppState(filepath.Join(w.dir, whatsappStateFileName))
+ if err != nil {
+ return nil, err
+ }
+
+ w.inbox = newWhatsAppInbox(w.state, clock, w.groups, &ImageStore{dir: filepath.Join(dir, imageDirName)}, logger)
+
+ info, err := os.Stat(w.sessionPath())
+ if errors.Is(err, os.ErrNotExist) {
+ return w, nil
+ }
+
+ if err != nil {
+ return nil, fmt.Errorf("read WhatsApp session: %w", err)
+ }
+
+ if info.Mode().Perm()&0o077 != 0 {
+ return nil, fmt.Errorf("%s holds the WhatsApp session, so its mode must be 0600", filepath.Join(whatsappDirName, whatsappSessionFileName))
+ }
+
+ err = w.openSession(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ if w.device.ID != nil {
+ w.linked = true
+ w.inbox.waitForCatchup()
+
+ w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
+ }
+
+ return w, nil
+}
+
// Run connects a linked device, logs the closed batches in l, runs the pair and
// unlink commands, and handles a logout from the phone, until ctx ends. Its
// notes go to the log of l, and notify gets the notices for all clients. A
return whatsappRedact(w.own, text)
}
-// WhatsAppLink is a copy of the link of the device at one time, for the parts
-// of WhatsApp that do not own it: the groups, the batches, find, and send.
-type WhatsAppLink struct {
- linked bool
- device *store.Device
- client WhatsAppClient
- connected bool
- own []string
-}
-
-// WhatsAppRequest is a pair or unlink command for Run, with the function that
-// gives its answers to the client that asked. A pairing has the phone number.
-type WhatsAppRequest struct {
- phone string
- unlink bool
- answer func(text string)
-}
-
// whatsappSetDeviceProps sets the device properties that the library sends at
// the pairing. They are global values of the library, so the daemon sets them
// once before any client exists. The phone shows the linked device with the
Messages []WhatsAppMessage `json:"messages"`
}
+// WhatsAppChat is a chat: its ID, its name, and whether it is a group.
+type WhatsAppChat struct {
+ ID string `json:"id"`
+ Name string `json:"name"`
+ Group bool `json:"group,omitempty"`
+}
+
+// WhatsAppMessage is one message of a batch: its ID, the name of its sender,
+// its WhatsApp time, and its text. Mine marks a message of the user. Mention
+// and Reply mark a message that mentions the user or replies to a message of
+// the user.
+type WhatsAppMessage struct {
+ ID string `json:"id"`
+ Sender string `json:"sender"`
+ Time time.Time `json:"time"`
+ Text string `json:"text"`
+ Mine bool `json:"mine,omitempty"`
+ Mention bool `json:"mention,omitempty"`
+ Reply bool `json:"reply,omitempty"`
+}
+
// Turn tells if the batch starts a turn: a private batch with a message of
// another person, or a group batch with a message of another person that
// mentions the user or replies to a message of the user.
return strings.Join(lines, "\n")
}
-
-// WhatsAppChat is a chat: its ID, its name, and whether it is a group.
-type WhatsAppChat struct {
- ID string `json:"id"`
- Name string `json:"name"`
- Group bool `json:"group,omitempty"`
-}
-
-// WhatsAppMessage is one message of a batch: its ID, the name of its sender,
-// its WhatsApp time, and its text. Mine marks a message of the user. Mention
-// and Reply mark a message that mentions the user or replies to a message of
-// the user.
-type WhatsAppMessage struct {
- ID string `json:"id"`
- Sender string `json:"sender"`
- Time time.Time `json:"time"`
- Text string `json:"text"`
- Mine bool `json:"mine,omitempty"`
- Mention bool `json:"mention,omitempty"`
- Reply bool `json:"reply,omitempty"`
-}
connections []context.Context
}
+// WhatsAppFakeSend is one send of the fake: the chat, the text, and the message
+// ID.
+type WhatsAppFakeSend struct {
+ to string
+ text string
+ id string
+}
+
+// WhatsAppRetryTest runs WhatsApp without a model or network client call. All
+// timers and channels belong to its synctest bubble, so backoff uses fake time.
+type WhatsAppRetryTest struct {
+ whatsapp *WhatsApp
+ fake *WhatsAppFake
+ logs *SyncBuffer
+ loop *TurnLoop
+ events *TurnEventRecorder
+ cancel context.CancelFunc
+ done chan error
+}
+
// ConnectContext keeps ctx and tells the test that the daemon connected.
func (f *WhatsAppFake) ConnectContext(ctx context.Context) error {
f.mutex.Lock()
return f
}
-// WhatsAppFakeSend is one send of the fake: the chat, the text, and the message
-// ID.
-type WhatsAppFakeSend struct {
- to string
- text string
- id string
-}
-
-// WhatsAppRetryTest runs WhatsApp without a model or network client call. All
-// timers and channels belong to its synctest bubble, so backoff uses fake time.
-type WhatsAppRetryTest struct {
- whatsapp *WhatsApp
- fake *WhatsAppFake
- logs *SyncBuffer
- loop *TurnLoop
- events *TurnEventRecorder
- cancel context.CancelFunc
- done chan error
-}
-
func (r *WhatsAppRetryTest) start(t *testing.T) {
t.Helper()
whatsappNamesNotReady = "The names of the contacts and groups have not arrived from the phone yet. They come some minutes after the link. Try again later."
)
+// WhatsAppMatch is one contact or group of whatsappFind.
+type WhatsAppMatch struct {
+ id string
+ name string
+ kind string
+}
+
// whatsappFind returns the contacts and the groups whose name contains name,
// with no regard to case and accents, as lines "ID | name | contact" or "ID |
// name | group", sorted by name. It gives at most whatsappFindMatches lines and
return whatsappFindAnswer(name, matches, read || len(names) > 0), nil
}
-// WhatsAppMatch is one contact or group of whatsappFind.
-type WhatsAppMatch struct {
- id string
- name string
- kind string
-}
-
// whatsappFold returns text in lower case with no accents and no space at its
// ends, so that names match with no regard to case and accents.
func whatsappFold(text string) string {
whatsappGroupsAge = time.Hour
)
-// newWhatsAppGroups returns the names of the groups with no names yet. logger
-// gets the failures of a read.
-func newWhatsAppGroups(logger *slog.Logger) *WhatsAppGroups {
- return &WhatsAppGroups{
- logger: logger,
- names: make(map[string]string),
- }
-}
-
// WhatsAppGroups holds the names of the groups of the account, by the ID of the
// group. They come from the events of the groups and from reads of the groups
// from WhatsApp, and they stay in memory only. triedAt is the time of the last
readAt time.Time
}
+// newWhatsAppGroups returns the names of the groups with no names yet. logger
+// gets the failures of a read.
+func newWhatsAppGroups(logger *slog.Logger) *WhatsAppGroups {
+ return &WhatsAppGroups{
+ logger: logger,
+ names: make(map[string]string),
+ }
+}
+
// name returns the name of the group id, or whatsappUnknownGroup. A group that
// is not in the names reads the groups from WhatsApp with the client of link
// again, at most once in whatsappGroupsRefresh.
whatsappImportUnknown = regexp.MustCompile(`^\[WhatsApp\] (.*), [A-Za-z]+ (\d{4}-\d{2}-\d{2})$`)
)
+// WhatsAppImportPlan holds the new chat-day batches and the size of the merged
+// log. Its estimate uses the measured cost of 1.3 calls per message, not a
+// promise about a particular export or compaction retry.
+type WhatsAppImportPlan struct {
+ agent string
+ cutoffs map[string]time.Time
+ added []DatedMessage
+ chats []string
+ cutoff time.Time
+ first time.Time
+ last time.Time
+ bytes int
+ messages int
+ skipped int
+ ownerKnown bool
+}
+
+// WhatsAppImportOptions selects the target, exports, and plan-only mode.
+type WhatsAppImportOptions struct {
+ planOnly bool
+ agent string
+ paths []string
+}
+
+// Write prints the plan without chat text or the owner's profile name.
+func (p WhatsAppImportPlan) Write(out io.Writer) error {
+ cutoff := "none (no live WhatsApp batch)"
+ if !p.cutoff.IsZero() {
+ cutoff = p.cutoff.Format(time.RFC3339)
+ }
+
+ dateRange := "none"
+ if !p.first.IsZero() {
+ dateRange = p.first.Format(time.RFC3339) + " to " + p.last.Format(time.RFC3339)
+ }
+
+ owner := "known (mapped to You)"
+ if !p.ownerKnown {
+ owner = "unknown (sender names kept as written)"
+ }
+
+ _, err := fmt.Fprintf(out, "agent: %s\nchats: %d\n", p.agent, len(p.chats))
+ if err != nil {
+ return fmt.Errorf("write import plan: %w", err)
+ }
+
+ for _, chat := range p.chats {
+ line := " " + chat
+ cutoff, found := p.cutoffs[chat]
+ if found {
+ line += " (cutoff: " + cutoff.Format(time.RFC3339) + ")"
+ }
+
+ _, err = fmt.Fprintln(out, line)
+ if err != nil {
+ return fmt.Errorf("write import plan: %w", err)
+ }
+ }
+
+ estimate := 1.3 * float64(p.messages)
+ if len(p.added) == 0 {
+ estimate = 0
+ }
+
+ lines := []string{
+ fmt.Sprintf("batches: %d (%d chat-days skipped)", len(p.added), p.skipped),
+ "dates: " + dateRange,
+ "cutoff: " + cutoff,
+ "owner: " + owner,
+ fmt.Sprintf("bytes: %d (new batch text)", p.bytes),
+ fmt.Sprintf("messages: %d (merged log)", p.messages),
+ fmt.Sprintf("estimated calls: %.0f (about 1.3 per message, limit %d)", estimate, rebuildCallLimit),
+ }
+
+ _, err = io.WriteString(out, strings.Join(lines, "\n")+"\n")
+ if err != nil {
+ return fmt.Errorf("write import plan: %w", err)
+ }
+
+ return nil
+}
+
// whatsappImport runs the offline import command. It holds the same token lock
// as the daemon and probe, even for a plan. The plan reads no credentials for a
// model and makes no network call. Only a completed rebuild replaces the agent
return nil
}
-// WhatsAppImportPlan holds the new chat-day batches and the size of the merged
-// log. Its estimate uses the measured cost of 1.3 calls per message, not a
-// promise about a particular export or compaction retry.
-type WhatsAppImportPlan struct {
- agent string
- cutoffs map[string]time.Time
- added []DatedMessage
- chats []string
- cutoff time.Time
- first time.Time
- last time.Time
- bytes int
- messages int
- skipped int
- ownerKnown bool
-}
-
-// Write prints the plan without chat text or the owner's profile name.
-func (p WhatsAppImportPlan) Write(out io.Writer) error {
- cutoff := "none (no live WhatsApp batch)"
- if !p.cutoff.IsZero() {
- cutoff = p.cutoff.Format(time.RFC3339)
- }
-
- dateRange := "none"
- if !p.first.IsZero() {
- dateRange = p.first.Format(time.RFC3339) + " to " + p.last.Format(time.RFC3339)
- }
-
- owner := "known (mapped to You)"
- if !p.ownerKnown {
- owner = "unknown (sender names kept as written)"
- }
-
- _, err := fmt.Fprintf(out, "agent: %s\nchats: %d\n", p.agent, len(p.chats))
- if err != nil {
- return fmt.Errorf("write import plan: %w", err)
- }
-
- for _, chat := range p.chats {
- line := " " + chat
- cutoff, found := p.cutoffs[chat]
- if found {
- line += " (cutoff: " + cutoff.Format(time.RFC3339) + ")"
- }
-
- _, err = fmt.Fprintln(out, line)
- if err != nil {
- return fmt.Errorf("write import plan: %w", err)
- }
- }
-
- estimate := 1.3 * float64(p.messages)
- if len(p.added) == 0 {
- estimate = 0
- }
-
- lines := []string{
- fmt.Sprintf("batches: %d (%d chat-days skipped)", len(p.added), p.skipped),
- "dates: " + dateRange,
- "cutoff: " + cutoff,
- "owner: " + owner,
- fmt.Sprintf("bytes: %d (new batch text)", p.bytes),
- fmt.Sprintf("messages: %d (merged log)", p.messages),
- fmt.Sprintf("estimated calls: %.0f (about 1.3 per message, limit %d)", estimate, rebuildCallLimit),
- }
-
- _, err = io.WriteString(out, strings.Join(lines, "\n")+"\n")
- if err != nil {
- return fmt.Errorf("write import plan: %w", err)
- }
-
- return nil
-}
-
-// WhatsAppImportOptions selects the target, exports, and plan-only mode.
-type WhatsAppImportOptions struct {
- planOnly bool
- agent string
- paths []string
-}
-
// whatsappImportParse accepts GNU-style flags, including --agent=NAME and --
// before paths that start with a dash.
func whatsappImportParse(args []string) (WhatsAppImportOptions, error) {
whatsappCatchupTimeout = time.Minute
)
-// newWhatsAppInbox returns the inbox of the batches in state, with clock for
-// the times of the batch texts and groups for the names of the groups. images
-// holds shared images, and logger gets image failures. Audio shares the data
-// directory of state.
-func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *WhatsAppGroups, images *ImageStore, logger *slog.Logger) *WhatsAppInbox {
- return &WhatsAppInbox{
- state: state,
- clock: clock,
- groups: groups,
- images: images,
- voices: &VoiceStore{dir: filepath.Join(filepath.Dir(filepath.Dir(state.path)), voiceDirName)},
- voice: &VoiceTranscriber{},
- logger: logger,
- wake: make(chan struct{}, 1),
- failures: make(chan error, 1),
- }
-}
-
// WhatsAppInbox puts messages into batches, one chat at a time, as
// WhatsAppState says. deliver logs each closed batch as a work message that
// starts with [WhatsApp], and a batch that needs the user wakes the turn loop.
catchupTimer *time.Timer
}
+// newWhatsAppInbox returns the inbox of the batches in state, with clock for
+// the times of the batch texts and groups for the names of the groups. images
+// holds shared images, and logger gets image failures. Audio shares the data
+// directory of state.
+func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *WhatsAppGroups, images *ImageStore, logger *slog.Logger) *WhatsAppInbox {
+ return &WhatsAppInbox{
+ state: state,
+ clock: clock,
+ groups: groups,
+ images: images,
+ voices: &VoiceStore{dir: filepath.Join(filepath.Dir(filepath.Dir(state.path)), voiceDirName)},
+ voice: &VoiceTranscriber{},
+ logger: logger,
+ wake: make(chan struct{}, 1),
+ failures: make(chan error, 1),
+ }
+}
+
// due returns a channel that gets a value when the next open batch is due, or
// nil when no batch is open.
func (i *WhatsAppInbox) due() <-chan time.Time {
whatsappSendRecords = 1000
)
-// openWhatsAppState reads the batches and the send records of the state file at
-// path. A missing file holds none. A batch closes after
-// whatsappBatchIdleTimeout with no message, or after whatsappBatchTimeout at
-// the latest.
-func openWhatsAppState(path string) (*WhatsAppState, error) {
- s := &WhatsAppState{
- path: path,
- }
-
- data, err := os.ReadFile(path)
- if errors.Is(err, os.ErrNotExist) {
- return s, nil
- }
-
- if err != nil {
- return nil, fmt.Errorf("read WhatsApp state: %w", err)
- }
-
- err = json.Unmarshal(data, &s.file)
- if err != nil {
- return nil, fmt.Errorf("%s: %w", path, err)
- }
-
- return s, nil
-}
-
// WhatsAppState holds the state file: the send records, and the batches of the
// WhatsApp messages that are not in the log yet, open and closed, in the order
// of their first message. The messages of one chat form one batch while the
file WhatsAppStateFile
}
+// WhatsAppStateFile is the content of the state file: the number of the last
+// batch, the batches that are not in the log yet, and the send records, oldest
+// first.
+type WhatsAppStateFile struct {
+ Next int `json:"next"`
+ Batches []WhatsAppBatch `json:"batches"`
+ Sends []WhatsAppSend `json:"sends"`
+}
+
+// WhatsAppSend is the record of one send: the ID that the agent gave, the chat,
+// the SHA-256 of the text, an optional stored image name, the WhatsApp message
+// ID, the state, which is sending, sent, or failed, the answer of the tool, and
+// the time of its start.
+type WhatsAppSend struct {
+ ID string `json:"id"`
+ Chat string `json:"chat"`
+ Digest string `json:"digest"`
+ Image string `json:"image,omitempty"`
+ MessageID string `json:"message_id"`
+ State string `json:"state"`
+ Answer string `json:"answer,omitempty"`
+ Time time.Time `json:"time"`
+}
+
+// openWhatsAppState reads the batches and the send records of the state file at
+// path. A missing file holds none. A batch closes after
+// whatsappBatchIdleTimeout with no message, or after whatsappBatchTimeout at
+// the latest.
+func openWhatsAppState(path string) (*WhatsAppState, error) {
+ s := &WhatsAppState{
+ path: path,
+ }
+
+ data, err := os.ReadFile(path)
+ if errors.Is(err, os.ErrNotExist) {
+ return s, nil
+ }
+
+ if err != nil {
+ return nil, fmt.Errorf("read WhatsApp state: %w", err)
+ }
+
+ err = json.Unmarshal(data, &s.file)
+ if err != nil {
+ return nil, fmt.Errorf("%s: %w", path, err)
+ }
+
+ return s, nil
+}
+
// Add puts message of chat into the open batch of chat, or into a new batch,
// with arrival as its time of arrival. A message that the batch holds already,
// which the servers sent again, changes nothing. It returns true when a batch
return nil
}
-
-// WhatsAppStateFile is the content of the state file: the number of the last
-// batch, the batches that are not in the log yet, and the send records, oldest
-// first.
-type WhatsAppStateFile struct {
- Next int `json:"next"`
- Batches []WhatsAppBatch `json:"batches"`
- Sends []WhatsAppSend `json:"sends"`
-}
-
-// WhatsAppSend is the record of one send: the ID that the agent gave, the chat,
-// the SHA-256 of the text, an optional stored image name, the WhatsApp message
-// ID, the state, which is sending, sent, or failed, the answer of the tool, and
-// the time of its start.
-type WhatsAppSend struct {
- ID string `json:"id"`
- Chat string `json:"chat"`
- Digest string `json:"digest"`
- Image string `json:"image,omitempty"`
- MessageID string `json:"message_id"`
- State string `json:"state"`
- Answer string `json:"answer,omitempty"`
- Time time.Time `json:"time"`
-}
projects / onechat / commitdiff