+func TestNotifyCLIReachesOnlyTheRootThroughTheDaemon(t *testing.T) {
+ server := newModelFake(t)
+ socket := daemonStartTest(t, server)
+ t.Setenv("ONECHAT_DIR", filepath.Dir(socket))
+
+ var out bytes.Buffer
+
+ args := []string{"--source", "test-check", "--id", "event-1", "--text", "A result needs review."}
+ require.NoError(t, notify(t.Context(), args, &out))
+
+ var receipt Event
+
+ require.NoError(t, json.Unmarshal(out.Bytes(), &receipt))
+ require.Equal(t, "notify", receipt.Kind)
+ require.Equal(t, "accepted", receipt.Status)
+ require.Positive(t, receipt.ID)
+ require.NotContains(t, out.String(), "needs review")
+ require.Eventually(t, func() bool {
+ return server.Calls() == 1
+ }, 10*time.Second, time.Millisecond)
+
+ texts := turnInputTexts(t, server.Call(0))
+ require.Contains(t, texts[len(texts)-1], "Untrusted external data")
+ require.Contains(t, texts[len(texts)-1], `"source":"test-check"`)
+ require.Contains(t, texts[len(texts)-1], `"id":"event-1"`)
+
+ out.Reset()
+ require.NoError(t, notify(t.Context(), args, &out))
+
+ var duplicate Event
+
+ require.NoError(t, json.Unmarshal(out.Bytes(), &duplicate))
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ root, err := openAgent(filepath.Join(filepath.Dir(socket), daemonAgentDirName))
+ require.NoError(t, err)
+ out.Reset()
+ require.NoError(t, notify(t.Context(), append([]string{"--agent-id", root.config.ID}, args...), &out))
+ require.NoError(t, json.Unmarshal(out.Bytes(), &duplicate))
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ args[len(args)-1] = "Changed result."
+ require.ErrorContains(t, notify(t.Context(), args, &out), "conflicts")
+}
+
+func TestNotifyCLISelectsOnlyThePermanentChildByID(t *testing.T) {
+ directory := daemonTestSocketDir(t)
+ daemonChildConfig(t, directory, "Garden")
+ daemonChildConfig(t, directory, "Library")
+
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ agents, err := openDaemonAgents(t.Context(), directory, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
+ require.NoError(t, err)
+
+ whatsapp, _, _ := whatsappStartTest(t, directory, newTestClock())
+ listener, err := daemonListenSocket(filepath.Join(directory, socketFileName))
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 1)
+ go func() {
+ done <- daemonServe(ctx, listener, nil, whatsapp, agents[0].schedules, func(events func(Event)) (*TurnLoop, error) {
+ return agents[0].Loop(whatsapp, events)
+ }, agents[1:]...)
+ }()
+
+ t.Cleanup(func() {
+ cancel()
+ require.NoError(t, <-done)
+ require.NoError(t, agents.Close())
+ })
+
+ t.Setenv("ONECHAT_DIR", directory)
+
+ var out bytes.Buffer
+
+ args := []string{"--agent-id", agents[1].agent.config.ID, "--source", "check", "--id", "child-1", "--text", "A result for only the child."}
+ require.NoError(t, notify(t.Context(), args, &out))
+
+ var accepted Event
+
+ require.NoError(t, json.Unmarshal(out.Bytes(), &accepted))
+ require.Equal(t, "accepted", accepted.Status)
+ require.Eventually(t, func() bool {
+ data, err := os.ReadFile(filepath.Join(agents[0].agent.dir, "messages.json"))
+ if err != nil {
+ return false
+ }
+
+ var queue DaemonMessageFile
+
+ err = json.Unmarshal(data, &queue)
+
+ return err == nil && len(queue.Pending) == 0
+ }, 10*time.Second, time.Millisecond)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
+ require.Zero(t, agents[0].memory.Len())
+ require.Zero(t, agents[2].memory.Len())
+
+ out.Reset()
+ require.NoError(t, notify(t.Context(), args, &out))
+
+ var duplicate Event
+
+ require.NoError(t, json.Unmarshal(out.Bytes(), &duplicate))
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Equal(t, accepted.ID, duplicate.ID)
+
+ args[1] = agents[2].agent.config.ID
+ require.ErrorContains(t, notify(t.Context(), args, &out), "conflicts")
+
+ args[1] = strings.Repeat("f", 2*agentIDBytes)
+ require.ErrorContains(t, notify(t.Context(), args, &out), "identity is unknown")
+ require.Equal(t, 1, server.Calls())
+}
+
+func TestNotifyFlagsValidateBeforeConnecting(t *testing.T) {
+ testCases := []struct {
+ name string
+ args []string
+ }{
+ {
+ name: "requires all fields",
+ args: []string{"--source", "check", "--id", "1"},
+ },
+ {
+ name: "rejects short long flags",
+ args: []string{"-source", "check", "--id", "1", "--text", "data"},
+ },
+ {
+ name: "rejects name selectors",
+ args: []string{"--agent", "Garden", "--source", "check", "--id", "1", "--text", "data"},
+ },
+ {
+ name: "rejects an empty explicit agent ID",
+ args: []string{"--agent-id=", "--source", "check", "--id", "1", "--text", "data"},
+ },
+ {
+ name: "rejects malformed agent IDs",
+ args: []string{"--agent-id=Garden", "--source", "check", "--id", "1", "--text", "data"},
+ },
+ {
+ name: "rejects repeated agent IDs",
+ args: []string{"--agent-id", strings.Repeat("a", 2*agentIDBytes), "--agent-id", strings.Repeat("a", 2*agentIDBytes), "--source", "check", "--id", "1", "--text", "data"},
+ },
+ {
+ name: "rejects a flag without a value",
+ args: []string{"--source"},
+ },
+ {
+ name: "rejects a repeated flag",
+ args: []string{"--source", "check", "--source", "check", "--id", "1", "--text", "data"},
+ },
+ {
+ name: "bounds text bytes",
+ args: []string{"--source", "check", "--id", "1", "--text", strings.Repeat("x", notifyTextBytes+1)},
+ },
+ {
+ name: "rejects invalid text encoding",
+ args: []string{"--source", "check", "--id", "1", "--text", string([]byte{0xff})},
+ },
+ {
+ name: "rejects unbounded identifiers",
+ args: []string{"--source", strings.Repeat("x", notifyIDBytes+1), "--id", "1", "--text", "data"},
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Setenv("ONECHAT_DIR", t.TempDir())
+ require.ErrorContains(t, notify(t.Context(), tc.args, &bytes.Buffer{}), "parse notification")
+ })
+ }
+
+ input, err := notifyParse([]string{"--agent-id=" + strings.Repeat("a", 2*agentIDBytes), "--source=check", "--id=event.1", "--text=Ol\u00e1"})
+ require.NoError(t, err)
+ require.Equal(t, "check", input.Source)
+ require.Equal(t, "event.1", input.ID)
+ require.Equal(t, strings.Repeat("a", 2*agentIDBytes), input.AgentID)
+}
+
+func TestNotifyHelpAndOfflineDaemon(t *testing.T) {
+ t.Setenv("ONECHAT_DIR", t.TempDir())
+
+ var out bytes.Buffer
+
+ require.NoError(t, notify(t.Context(), []string{"--help"}, &out))
+ require.Contains(t, out.String(), "not processed")
+ require.ErrorContains(t, notify(t.Context(), []string{"--source", "check", "--id", "1", "--text", "data"}, &out), "running daemon")
+}
+
+func TestNotifyCancellationInterruptsAReceiptWait(t *testing.T) {
+ directory := daemonTestSocketDir(t)
+ t.Setenv("ONECHAT_DIR", directory)
+
+ listener, err := net.Listen("unix", filepath.Join(directory, socketFileName))
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 1)
+ go func() {
+ done <- notify(ctx, []string{"--source", "check", "--id", "1", "--text", "data"}, &bytes.Buffer{})
+ }()
+
+ conn, err := listener.Accept()
+ require.NoError(t, err)
+
+ var request SocketRequest
+
+ require.NoError(t, json.NewDecoder(conn).Decode(&request))
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ require.NoError(t, conn.Close())
+ require.NoError(t, listener.Close())
+}
+
projects / onechat / blobdiff