Order model and chat declarations

[onechat] / whatsapp_import.go

diff --git a/whatsapp_import.go b/whatsapp_import.go
index 23619157ac437de4074b1b8b1f87eefca9cf3513..0021bc8be5904d820375a70e5635eec349ee3143 100644 (file)
--- a/whatsapp_import.go
+++ b/whatsapp_import.go
@@ -34,24 +34,97 @@ var (
        whatsappImportUnknown = regexp.MustCompile(`^\[WhatsApp\] (.*), [A-Za-z]+ (\d{4}-\d{2}-\d{2})$`)
 )
 
        whatsappImportUnknown = regexp.MustCompile(`^\[WhatsApp\] (.*), [A-Za-z]+ (\d{4}-\d{2}-\d{2})$`)
 )
 

-// 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
-// directory, and unfinished subagents prevent either mode.
-func whatsappImport(ctx context.Context, args []string, out io.Writer) (err error) {
-       planOnly := len(args) > 0 && args[0] == "-n"
-       if planOnly {
-               args = args[1:]
+// 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)"

        }
 
        }
 

-       if len(args) == 0 {
-               return errors.New("usage: onechat import [-n] ZIP [ZIP...]")
+       _, 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 _, arg := range args {
-               if strings.HasPrefix(arg, "-") {
-                       return errors.New("usage: onechat import [-n] ZIP [ZIP...]")
+       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
+// directory. Unfinished subagents and pending target deliveries prevent either
+// mode.
+func whatsappImport(ctx context.Context, args []string, out io.Writer) (err error) {
+       options, err := whatsappImportParse(args)
+       if err != nil {
+               return fmt.Errorf("parse import arguments: %w", err)

        }
 
        dir, err := dataDir()
        }
 
        dir, err := dataDir()


@@ -72,47 +145,57 @@ func whatsappImport(ctx context.Context, args []string, out io.Writer) (err erro
        }()
 
        root := filepath.Join(dir, "agent")
        }()
 
        root := filepath.Join(dir, "agent")

-       err = rebuildIdle(root)
+       agent, err := whatsappImportAgent(root, options.agent)
+       if err != nil {
+               return fmt.Errorf("select import agent: %w", err)
+       }
+
+       err = whatsappImportIdle(root, agent)

        if err != nil {
                return fmt.Errorf("check import work: %w", err)
        }
 
        if err != nil {
                return fmt.Errorf("check import work: %w", err)
        }
 

-       old, err := rebuildReadLog(ctx, root)
+       old, err := rebuildReadLog(ctx, agent.dir)

        if err != nil {
                return fmt.Errorf("read import log: %w", err)
        }
 
        if err != nil {
                return fmt.Errorf("read import log: %w", err)
        }
 

+       var rootLog []DatedMessage
+
+       if agent.dir != root {
+               rootLog, err = rebuildReadLog(ctx, root)
+               if err != nil {
+                       return fmt.Errorf("read root import history: %w", err)
+               }
+       }
+

        names, owner, err := whatsappImportNames(ctx, dir)
        if err != nil {
                return fmt.Errorf("read import names: %w", err)
        }
 
        names, owner, err := whatsappImportNames(ctx, dir)
        if err != nil {
                return fmt.Errorf("read import names: %w", err)
        }
 

-       plan, err := whatsappImportPlan(ctx, args, old, nil, names, owner, time.Local)
+       plan, err := whatsappImportPlan(ctx, options.paths, old, rootLog, names, owner, time.Local)

        if err != nil {
                return fmt.Errorf("plan import: %w", err)
        }
 
        if err != nil {
                return fmt.Errorf("plan import: %w", err)
        }
 

+       plan.agent = agent.config.Name

        err = plan.Write(out)
        if err != nil {
                return fmt.Errorf("write import plan: %w", err)
        }
 
        err = plan.Write(out)
        if err != nil {
                return fmt.Errorf("write import plan: %w", err)
        }
 

-       if planOnly || len(plan.added) == 0 {
+       if options.planOnly || len(plan.added) == 0 {

                return nil
        }
 
        // An existing agent is required so that a failed import never creates
        // or changes the original agent's configuration.
                return nil
        }
 
        // An existing agent is required so that a failed import never creates
        // or changes the original agent's configuration.

-       _, err = os.Stat(filepath.Join(root, "config.json"))
+       _, err = os.Stat(filepath.Join(agent.dir, "config.json"))

        if err != nil {
                return fmt.Errorf("read agent config: %w", err)
        }
 
        if err != nil {
                return fmt.Errorf("read agent config: %w", err)
        }
 

-       agent, err := openAgent(root)
-       if err != nil {
-               return fmt.Errorf("open import agent: %w", err)
-       }
-

        config, err := loadConfig(dir)
        if err != nil {
                return fmt.Errorf("load import configuration: %w", err)
        config, err := loadConfig(dir)
        if err != nil {
                return fmt.Errorf("load import configuration: %w", err)


@@ -142,7 +225,7 @@ func whatsappImport(ctx context.Context, args []string, out io.Writer) (err erro
        }
 
        client := newModelClient(address, auth, agent.CacheKey(), usagePath)
        }
 
        client := newModelClient(address, auth, agent.CacheKey(), usagePath)

-       backup, err := rebuild(ctx, agent, plan.added, client, newCompactor(client, agent.Prompt, config.Compactor, agent.Tools()), out)
+       backup, err := rebuild(ctx, agent, plan.added, client, newCompactor(client, agent.Prompt, config.Compactor, agent.Tools(), agent.Images()), out)

        if err != nil {
                return fmt.Errorf("rebuild imported memory: %w", err)
        }
        if err != nil {
                return fmt.Errorf("rebuild imported memory: %w", err)
        }


@@ -155,75 +238,128 @@ func whatsappImport(ctx context.Context, args []string, out io.Writer) (err erro
        return nil
 }
 
        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 {
-       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
+// whatsappImportParse accepts GNU-style flags, including --agent=NAME and --
+// before paths that start with a dash.
+func whatsappImportParse(args []string) (WhatsAppImportOptions, error) {
+       var options WhatsAppImportOptions
+
+       flags := true
+       for i := 0; i < len(args); i++ {
+               arg := args[i]
+               switch {
+               case flags && arg == "--":
+                       flags = false
+               case flags && arg == "-n":
+                       options.planOnly = true
+               case flags && (arg == "--agent" || strings.HasPrefix(arg, "--agent=")):
+                       _, name, found := strings.Cut(arg, "=")
+                       if !found {
+                               i++
+                               if i == len(args) {
+                                       return options, errors.New("--agent needs a child name")
+                               }
+
+                               name = args[i]
+                       }
+
+                       if name == "" {
+                               return options, errors.New("--agent needs a child name")
+                       }
+
+                       options.agent = name
+               case flags && strings.HasPrefix(arg, "-"):
+                       return options, fmt.Errorf("unknown import flag %q", arg)
+               default:
+                       options.paths = append(options.paths, arg)
+               }
+       }
+
+       if len(options.paths) == 0 {
+               return options, errors.New("usage: onechat import [-n] [--agent NAME] ZIP [ZIP...]")
+       }
+
+       return options, nil

 }
 
 }
 

-// 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)
+// whatsappImportAgent selects the root or an existing direct child without
+// creating a directory or config. A root without a config can have a plan,
+// using the default name.
+func whatsappImportAgent(root, name string) (*Agent, error) {
+       _, err := os.Stat(filepath.Join(root, "config.json"))
+       if errors.Is(err, os.ErrNotExist) && name == "" {
+               var config AgentConfig
+
+               config, err = defaultAgentConfig()
+               if err != nil {
+                       return nil, fmt.Errorf("read default import agent: %w", err)
+               }
+
+               return &Agent{
+                       dir:    root,
+                       config: config,
+               }, nil

        }
 
        }
 

-       dateRange := "none"
-       if !p.first.IsZero() {
-               dateRange = p.first.Format(time.RFC3339) + " to " + p.last.Format(time.RFC3339)
+       if err != nil {
+               return nil, fmt.Errorf("read root agent config: %w", err)

        }
 
        }
 

-       owner := "known (mapped to You)"
-       if !p.ownerKnown {
-               owner = "unknown (sender names kept as written)"
+       agent, err := openAgent(root)
+       if err != nil {
+               return nil, fmt.Errorf("open root agent: %w", err)

        }
 
        }
 

-       _, err := fmt.Fprintf(out, "chats: %d\n", len(p.chats))
+       if name == "" {
+               return agent, nil
+       }
+
+       if agent.config.Parent != "" {
+               return nil, errors.New("import root cannot have a parent")
+       }
+
+       children, err := agent.Children()

        if err != nil {
        if err != nil {

-               return fmt.Errorf("write import plan: %w", err)
+               return nil, fmt.Errorf("read import children: %w", err)

        }
 
        }
 

-       for _, chat := range p.chats {
-               line := "  " + chat
-               cutoff, found := p.cutoffs[chat]
-               if found {
-                       line += " (cutoff: " + cutoff.Format(time.RFC3339) + ")"
+       for _, child := range children {
+               if child.config.Name == name {
+                       return child, nil

                }
                }

+       }

 
 

-               _, err = fmt.Fprintln(out, line)
-               if err != nil {
-                       return fmt.Errorf("write import plan: %w", err)
-               }
+       return nil, fmt.Errorf("child %q does not exist", name)
+}
+
+// whatsappImportIdle refuses unfinished subagents and pending deliveries to the
+// target. Rebuilding renumbers its log, so saved delivery IDs must not refer to
+// it until their turns finish. The caller holds the token lock.
+func whatsappImportIdle(root string, agent *Agent) error {
+       err := rebuildIdle(agent.dir)
+       if err != nil {
+               return fmt.Errorf("check target work: %w", err)

        }
 
        }
 

-       estimate := 1.3 * float64(p.messages)
-       if len(p.added) == 0 {
-               estimate = 0
+       data, err := os.ReadFile(filepath.Join(root, "messages.json"))
+       if errors.Is(err, os.ErrNotExist) {
+               return nil

        }
 
        }
 

-       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),
+       if err != nil {
+               return fmt.Errorf("read agent deliveries: %w", err)

        }
 
        }
 

-       _, err = io.WriteString(out, strings.Join(lines, "\n")+"\n")
+       var queue DaemonMessageFile
+
+       err = json.Unmarshal(data, &queue)

        if err != nil {
        if err != nil {

-               return fmt.Errorf("write import plan: %w", err)
+               return fmt.Errorf("decode agent deliveries: %w", err)
+       }
+
+       for _, message := range queue.Pending {
+               if message.Target == agent.config.Name {
+                       return fmt.Errorf("import refused: agent %q has pending deliveries", agent.config.Name)
+               }

        }
 
        return nil
        }
 
        return nil