From: Anderson A. Giacomolli Date: Wed, 11 Nov 2026 13:51:58 +0000 (-0300) Subject: Order model and chat declarations X-Git-Url: https://git.giacomolli.uk/onechat/commitdiff_plain/80e348724041a694002fc38f57e8868b9b9a2be9?ds=sidebyside Order model and chat declarations Model, memory, chat, tool, and view files put constants and variables before types, then constructors, grouped methods, and helpers. Shared test helpers use the same order, and ordinary tests keep tests first. Declaration text and function bodies stay unchanged. Issue: 947bddb --- diff --git a/clock_helper_test.go b/clock_helper_test.go index 32dfbb9..0a14fd8 100644 --- a/clock_helper_test.go +++ b/clock_helper_test.go @@ -7,12 +7,12 @@ type TestClock struct { 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 +} diff --git a/memory.go b/memory.go index 1b734c0..df99a95 100644 --- a/memory.go +++ b/memory.go @@ -22,6 +22,64 @@ var ( 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. @@ -75,46 +133,6 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar 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 { @@ -779,21 +797,3 @@ func (m *Memory) buildBuiltByModel(key NodeKey) (bool, 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"` -} diff --git a/model.go b/model.go index cc77f9a..074ff60 100644 --- a/model.go +++ b/model.go @@ -40,87 +40,6 @@ const ( // 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 { @@ -130,13 +49,6 @@ 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 @@ -161,6 +73,163 @@ type ModelClient struct { 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 @@ -278,13 +347,6 @@ func (c *ModelClient) send(httpRequest *http.Request, record *ModelUsageRecord) 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) @@ -312,133 +374,71 @@ func (u *ModelUsageLog) Write(record ModelUsageRecord) (err error) { 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, diff --git a/model_fake_helper_test.go b/model_fake_helper_test.go index 00fa808..f9d1451 100644 --- a/model_fake_helper_test.go +++ b/model_fake_helper_test.go @@ -34,6 +34,21 @@ type ModelFake struct { 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() @@ -80,21 +95,6 @@ func (f *ModelFake) serve(w http.ResponseWriter, r *http.Request) { 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", diff --git a/probe.go b/probe.go index 384d0f8..a1ee38d 100644 --- a/probe.go +++ b/probe.go @@ -23,6 +23,28 @@ const ( 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. @@ -145,28 +167,6 @@ func probe(ctx context.Context, args []string, out io.Writer) (err error) { 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) { diff --git a/probe_log_helper_test.go b/probe_log_helper_test.go index 0097dda..7df9f48 100644 --- a/probe_log_helper_test.go +++ b/probe_log_helper_test.go @@ -10,6 +10,21 @@ import ( // 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 @@ -133,18 +148,3 @@ func probeLog(count int) []DatedMessage { 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"} -) diff --git a/prompt.go b/prompt.go index 0852bec..29e58ee 100644 --- a/prompt.go +++ b/prompt.go @@ -34,9 +34,6 @@ const ( 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 @@ -197,6 +194,9 @@ date(id) gives the date and time of message id. zoom("Name") gives an agent's 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 diff --git a/subagent.go b/subagent.go index 4cb6381..035f57d 100644 --- a/subagent.go +++ b/subagent.go @@ -23,19 +23,6 @@ const ( 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 @@ -50,6 +37,30 @@ type Subagents struct { 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) { @@ -208,17 +219,6 @@ func (s *Subagents) Close() 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() diff --git a/subagent_run.go b/subagent_run.go index faa6207..3df12ce 100644 --- a/subagent_run.go +++ b/subagent_run.go @@ -78,6 +78,25 @@ type SubagentRuns struct { 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. @@ -551,25 +570,6 @@ func (r *SubagentRuns) take(run *SubagentRun) []Message { 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. diff --git a/summarizer_helper_test.go b/summarizer_helper_test.go index e78a69e..6544b74 100644 --- a/summarizer_helper_test.go +++ b/summarizer_helper_test.go @@ -21,6 +21,10 @@ type FakeSummarizer struct { tasks []Task } +func newFakeSummarizer() *FakeSummarizer { + return &FakeSummarizer{} +} + func (f *FakeSummarizer) Summarize(ctx context.Context, task Task) (string, error) { f.mutex.Lock() @@ -90,7 +94,3 @@ func fakeLine(task Task) string { return line } - -func newFakeSummarizer() *FakeSummarizer { - return &FakeSummarizer{} -} diff --git a/syncer.go b/syncer.go index 3da6d7a..0784393 100644 --- a/syncer.go +++ b/syncer.go @@ -10,22 +10,6 @@ import ( "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, @@ -48,6 +32,30 @@ type Syncer struct { 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() { @@ -138,11 +146,3 @@ func (s *Syncer) run(target uint64) { 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{}), - } -} diff --git a/syncer_test.go b/syncer_test.go index 16f3345..d463eec 100644 --- a/syncer_test.go +++ b/syncer_test.go @@ -148,6 +148,13 @@ type SyncerGatedSync 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() @@ -184,13 +191,6 @@ func (g *SyncerGatedSync) Count() int { 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. diff --git a/telegram.go b/telegram.go index 4ed0861..59ea9b4 100644 --- a/telegram.go +++ b/telegram.go @@ -58,6 +58,66 @@ const ( 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) { @@ -116,13 +176,6 @@ func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *sl } } -// 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. @@ -139,36 +192,6 @@ func (c TelegramConfig) validate() error { 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. @@ -589,17 +612,6 @@ func (b *TelegramBot) notice(text string) { }) } -// 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() { @@ -621,18 +633,6 @@ func (o *TelegramOutage) End() { 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) { diff --git a/telegram_api.go b/telegram_api.go index 6967c99..3a44a0d 100644 --- a/telegram_api.go +++ b/telegram_api.go @@ -49,50 +49,113 @@ const ( 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 @@ -368,102 +431,39 @@ func (b *TelegramAPI) redact(err error) string { 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) } diff --git a/telegram_fake_helper_test.go b/telegram_fake_helper_test.go index b71eee0..6082d7d 100644 --- a/telegram_fake_helper_test.go +++ b/telegram_fake_helper_test.go @@ -50,6 +50,42 @@ type TelegramFake struct { 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() @@ -283,42 +319,6 @@ func (f *TelegramFake) from(offset int64) []TelegramUpdate { 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{ diff --git a/telegram_markdown.go b/telegram_markdown.go index fc68634..b6ea5a2 100644 --- a/telegram_markdown.go +++ b/telegram_markdown.go @@ -14,6 +14,75 @@ var ( 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 @@ -102,75 +171,6 @@ func telegramUTF16Units(text string) int { 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. diff --git a/tools.go b/tools.go index 621cd84..11e208d 100644 --- a/tools.go +++ b/tools.go @@ -72,6 +72,113 @@ type Toolbox struct { 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 @@ -493,113 +600,6 @@ func (b *Toolbox) path(path string) string { 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 { diff --git a/tools_test.go b/tools_test.go index 487a5fc..e1b19bb 100644 --- a/tools_test.go +++ b/tools_test.go @@ -223,6 +223,26 @@ type ToolRecordingLog struct { 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() @@ -253,23 +273,3 @@ func (r *ToolRecordingLog) Events() []string { 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, - } -} diff --git a/tree.go b/tree.go index 079969c..ad30eeb 100644 --- a/tree.go +++ b/tree.go @@ -23,6 +23,44 @@ const nodeBytes = 512 // 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) { @@ -50,70 +88,6 @@ func openTree(ctx context.Context, root string, total int, clock func() time.Tim 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() @@ -393,21 +367,47 @@ func (t *Tree) Count() int { 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 diff --git a/turn.go b/turn.go index 9e6a1a7..b5aa4cf 100644 --- a/turn.go +++ b/turn.go @@ -10,6 +10,54 @@ import ( "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 @@ -70,54 +118,6 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W 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. diff --git a/turn_test.go b/turn_test.go index 811487e..733c9a9 100644 --- a/turn_test.go +++ b/turn_test.go @@ -263,6 +263,59 @@ type TurnTestLoop struct { 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() @@ -279,12 +332,6 @@ func (l *TurnTestLoop) Log(t *testing.T) []string { 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() @@ -343,12 +390,6 @@ func (r *TurnEventRecorder) Add(event Event) { 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() @@ -383,47 +424,6 @@ func turnInputTexts(t *testing.T, call ModelFakeCall) []string { 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", diff --git a/view.go b/view.go index 2357198..1b3ed6a 100644 --- a/view.go +++ b/view.go @@ -23,42 +23,6 @@ const ( 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< 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 @@ -83,6 +47,27 @@ type View struct { 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{ @@ -343,14 +328,29 @@ func (v *View) lineSize(line NodeKey) int { 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< 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 } diff --git a/view_layout.go b/view_layout.go index 3db2f01..a2d4825 100644 --- a/view_layout.go +++ b/view_layout.go @@ -14,6 +14,30 @@ import ( // 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 "\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) { @@ -38,23 +62,6 @@ 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 "\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. @@ -103,13 +110,6 @@ func (l *ViewLayout) Split(lines []string) ([]string, error) { 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 { diff --git a/view_test.go b/view_test.go index 46f76da..f771eb2 100644 --- a/view_test.go +++ b/view_test.go @@ -149,14 +149,6 @@ func TestRestoreViewRejectsAGap(t *testing.T) { // 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. @@ -166,6 +158,19 @@ type ViewPushEntry struct { 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 @@ -179,11 +184,6 @@ func (e *ViewPushEntry) 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 diff --git a/whatsapp.go b/whatsapp.go index 4950d28..ee785a3 100644 --- a/whatsapp.go +++ b/whatsapp.go @@ -98,60 +98,6 @@ type WhatsAppClient interface { 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 @@ -212,6 +158,78 @@ type WhatsApp struct { 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 @@ -1047,24 +1065,6 @@ func (w *WhatsApp) redact(text string) string { 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 diff --git a/whatsapp_batch.go b/whatsapp_batch.go index 7f43a3c..dcdd712 100644 --- a/whatsapp_batch.go +++ b/whatsapp_batch.go @@ -27,6 +27,27 @@ type WhatsAppBatch struct { 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. @@ -74,24 +95,3 @@ func (b WhatsAppBatch) Text(location *time.Location) string { 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"` -} diff --git a/whatsapp_fake_helper_test.go b/whatsapp_fake_helper_test.go index e5dc502..24789a7 100644 --- a/whatsapp_fake_helper_test.go +++ b/whatsapp_fake_helper_test.go @@ -96,6 +96,26 @@ type WhatsAppFake struct { 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() @@ -321,26 +341,6 @@ func (f *WhatsAppFake) newClient(_ *store.Device, handle func(event any) bool) W 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() diff --git a/whatsapp_find.go b/whatsapp_find.go index 9e3cbf8..5c19ae7 100644 --- a/whatsapp_find.go +++ b/whatsapp_find.go @@ -23,6 +23,13 @@ const ( 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 @@ -94,13 +101,6 @@ func whatsappFind(ctx context.Context, link WhatsAppLink, groups *WhatsAppGroups 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 { diff --git a/whatsapp_groups.go b/whatsapp_groups.go index 6e9bed3..5eedd5d 100644 --- a/whatsapp_groups.go +++ b/whatsapp_groups.go @@ -25,15 +25,6 @@ const ( 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 @@ -48,6 +39,15 @@ type WhatsAppGroups struct { 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. diff --git a/whatsapp_import.go b/whatsapp_import.go index 1092b49..0021bc8 100644 --- a/whatsapp_import.go +++ b/whatsapp_import.go @@ -34,6 +34,88 @@ var ( 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 @@ -156,88 +238,6 @@ func whatsappImport(ctx context.Context, args []string, out io.Writer) (err erro 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) { diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go index c9ca81b..a104a0d 100644 --- a/whatsapp_inbox.go +++ b/whatsapp_inbox.go @@ -32,24 +32,6 @@ const ( 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. @@ -90,6 +72,24 @@ type WhatsAppInbox struct { 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 { diff --git a/whatsapp_state.go b/whatsapp_state.go index 3d21b16..825fa18 100644 --- a/whatsapp_state.go +++ b/whatsapp_state.go @@ -31,32 +31,6 @@ const ( 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 @@ -83,6 +57,56 @@ type WhatsAppState struct { 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 @@ -287,27 +311,3 @@ func (s *WhatsAppState) save() error { 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"` -}