Feat/multiworker

src/AppBase.h

@@ -1,13 +1,15 @@
-#include <memory>
 #pragma once
+#include <memory>
 #include "AUI/Common/AObject.h"
 #include "AUI/Common/ATimer.h"
 #include "AUI/Thread/AAsyncHolder.h"
 #include "AUI/Thread/AEventLoop.h"
 #include "Diary.h"
 #include "IOpenAIChat.h"
 #include "MetricsBreadcumbs.h"
+#include "NotificationManager.h"
 #include "OpenAITools.h"
+#include "Worker.h"
 
 class AppBase : public AObject {
 public:
@@ -16,35 +18,14 @@ class AppBase : public AObject {
         _<IOpenAIChat> openAI;
     };
     AppBase(Init init);
-    virtual ~AppBase();
-    AString getSystemPrompt() const;
-
-    struct Notification {
-        AString message;
-        OpenAITools actions;
-        AFuture<> onStartedProcessing;
-        AFuture<> onProcessed;
-    };
-
-
-    /**
-     * @brief Passes an event to the AI to process
-     * @param notification notification text message in natural language (i.e., "you received a message from "...": ...;
-     * an
-     * @param actions immediate actions (tools) related to the notification (i.e., open related chat)
-     * alarm triggerred, etc...)
-     * @return Promise satisfied when the notification is processed.
-     * @details
-     * Think of it as your phone's notifications: you receive a notification, read it and (maybe) react to it.
-     */
-    const Notification& passNotificationToAI(AString notification, OpenAITools actions = {}, bool first = false);
+    virtual ~AppBase() = default;
+    AString getSystemPrompt();
 
     AFuture<> diaryDumpMessages();
+    AFuture<> diaryDumpMessages(IOpenAIChat::Session& temporaryContext);
 
     void actProactively();
 
-    [[nodiscard]] const IOpenAIChat::Session& temporaryContext() const { return mTemporaryContext; }
-
     [[nodiscard]] Diary& diary() { return mDiary; }
 
     /**
@@ -64,28 +45,22 @@ class AppBase : public AObject {
         return mMetricBreadcumbs;
     }
 
-    /**
-     * @brief If Kuni is sleeping, this function wake ups her.
-     */
-    void wakeUpIfSleeping() {
-        mWakeup = true;
-    }
-
-protected:
-    AAsyncHolder mAsync;
-    aui::float_within_0_1 mRelevanceThreshold = 0.5f;
-
-    // Set by llmuiOpenTelegramChat; read by updateTools to populate ToolCallEvent.
-    AOptional<std::chrono::system_clock::time_point> mLastOpenedChatLastMessageTime;
-
-    virtual AFuture<AString> onCleanContext();
+    virtual AString onCleanContext() const;
 
+    NotificationManager& notificationManager() {
+        return mNotificationManager;
+    }
 
     /**
      * @brief Called by the main coroutine when a notification was processed.
      */
     virtual void onOffline() {}
 
+    [[nodiscard]]
+    const _<IOpenAIChat>& openAI() const noexcept {
+        return mInit.openAI;
+    }
+
     /**
      * @brief Called by the main coroutine during LLM inference streaming to observe which tool calls LLM is about to
      * perform.
@@ -94,58 +69,27 @@ class AppBase : public AObject {
      */
     virtual void onResponseAssembling(IOpenAIChat::Response response) {}
 
-    /**
-     *
-     * @return @brief Called before LLM's processing loop.
-     */
-    virtual AFuture<> onBeforeMainLoop() { co_return; }
-
     /**
      * @brief Adds always available actions
      */
-    virtual void updateTools(OpenAITools& actions);
+    virtual void updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext);
 
-    /**
-     * @brief Removes notifications by the given substring.
-     * @param substring to search in notification texts. Must be unique enough to avoid false positives.
-     * @details
-     * Can be used to remove obsolete notifications from AI's queue.
-     */
-    void removeNotifications(const AString& substring);
 
-    IOpenAIChat::Session mTemporaryContext = [] {
-        IOpenAIChat::Session s;
-        s.sessionId = "kuni_main_coro";
-        return s;
-    }();
+    void wakeUpIfSleeping();
 
-    /**
-     * @brief Performs needed adjustments to the diary page and removes the page from listing. Formatted contents are
-     * returned.
-     * @details
-     * Adjusts usage count, last used and score fields, according to relatedness.
-     *
-     * Format's with XML tag with needed attributes.
-     *
-     * The diary page with new metadata is dropped onto disk and removes from mDiary. This ensures this specific diary
-     * page wouldn't be considered and included again until mTemporary context is cleaned via diaryDumpMessages.
-     *
-     */
-    [[nodiscard]]
-    AString takeDiaryEntry(const Diary::EntryExAndRelatedness& i);
+protected:
+    AAsyncHolder mAsync;
+
+    // Set by llmuiOpenTelegramChat; read by updateTools to populate ToolCallEvent.
+    AOptional<std::chrono::system_clock::time_point> mLastOpenedChatLastMessageTime;
 
-    [[nodiscard]]
-    const _<IOpenAIChat>& openAI() const noexcept {
-        return mInit.openAI;
-    }
 
 private:
     _<MetricsBreadcumbs> mMetricBreadcumbs = _new<MetricsBreadcumbs>();
     const Init mInit;
-    std::deque<Notification> mNotifications;
     AFuture<> mNotificationsSignal;
     _<ATimer> mWakeupTimer;
-    // OpenAITools mTools;
+    NotificationManager mNotificationManager;
     AString mSystemPromptSuffix;
 
     bool mWakeup = false;
@@ -161,6 +105,9 @@ class AppBase : public AObject {
      */
     bool mAskCalledThisTurn = false;
 
+    AVector<AArc<Worker>> mWorkers;
+    AProperty<size_t> mWorkerCount;
+    ASpinlockMutex mWorkingMemoryLock;
+
     Diary mDiary;
-    std::shared_ptr<bool> mAliveToken = std::make_shared<bool>(true);
 };

src/Diary.h

@@ -28,7 +28,7 @@
  */
 class Diary {
 public:
-    virtual ~Diary() = default;
+    ~Diary() = default;
     /**
      * @brief Simple representation of a diary entry.
      *
@@ -146,6 +146,9 @@ class Diary {
      */
     Diary(Init init);
 
+    Diary(const Diary&) = default;
+    Diary(Diary&&) noexcept = default;
+
     /**
      * @brief Persist a simple entry to disk.
      *
@@ -159,7 +162,7 @@ class Diary {
      * The metadata block is serialized to JSON and written before the
      * freeform body, surrounded by `---` delimiters.
      */
-    virtual void save(const EntryEx& entry);
+    void save(const EntryEx& entry);
 
     /**
      * @brief Remove an entry from the in‑memory cache.
@@ -184,7 +187,7 @@ class Diary {
      * [0,1], and returns a sorted vector of {@link EntryExAndRelatedness}
      * objects.
      */
-    virtual AFuture<AVector<EntryExAndRelatedness>> query(const std::valarray<double>& query, QueryOpts opts);
+    AFuture<AVector<EntryExAndRelatedness>> query(const std::valarray<double>& query, QueryOpts opts);
 
     /**
      * @brief Compute the relatedness of a single entry to a context vector.
@@ -232,16 +235,6 @@ class Diary {
 private:
     const Init mInit;
 
-    /**
-     * @brief Path to the directory containing the markdown files.
-     */
-
-    /**
-     * @brief Holds asynchronous tasks for the diary.
-     */
-
-    AAsyncHolder mAsync;
-
     /**
      * @brief Lazily cached list of parsed diary entries.
      */

src/ImageGenerator.cpp

@@ -189,17 +189,25 @@ AFuture<> ImageGenerator::engineerPrompt(PromptPair& out, const AString& descrip
             }
         };
     }();
+    naxyi_before:
     auto response = co_await mOpenAI->chat(params, messages);
     naxyi:
     if (response.choices.empty()) {
         throw AException("OpenAI returned no choices for initial prompt engineering");
     }
     auto content = response.choices[0].message.content;
-    auto json = parseResponse(content);
-    out = {
-        .positive = json["positivePrompt"].asString(),
-        .negative = json["negativePrompt"].asString(),
-    };
+    try {
+        auto json = parseResponse(content);
+        out = {
+            .positive = json["positivePrompt"].asString(),
+            .negative = json["negativePrompt"].asString(),
+        };
+    } catch (const AException& e) {
+        if (e.getMessage().contains("unexpected end of json stream")) {
+            goto naxyi_before;
+        }
+        throw;
+    }
 
     for (const auto&[name, prompt] : std::array {std::make_pair("positive", &out.positive), std::make_pair("negative", &out.negative) }) {
         prompt->replaceAll(") ", "), "); // add commas
@@ -247,6 +255,7 @@ AFuture<ImageGenerator::AssessmentResult> ImageGenerator::assessImage(const AIma
             .content = "Assess this image: " + IOpenAIChat::embedImage(image)
         }
     };
+    tryAgain:
     auto response = co_await mOpenAI->chat(params, messages);
 
     if (response.choices.empty()) {
@@ -262,6 +271,9 @@ AFuture<ImageGenerator::AssessmentResult> ImageGenerator::assessImage(const AIma
         };
         co_return result;
     } catch (const AException& e) {
+        if (e.getMessage().contains("unexpected end of json stream")) {
+            goto tryAgain;
+        }
         ALogger::err(LOG_TAG) << "Failed to parse assessment JSON: " << e << "\nContent: " << responseContent;
         // Fallback: assume satisfied if parsing fails to avoid infinite loops, but log error
         co_return AssessmentResult{.satisfied = false, .feedback = "" };

src/NotificationManager.cpp

@@ -0,0 +1,76 @@
+//
+// Created by alex2772 on 7/14/26.
+//
+
+#include "NotificationManager.h"
+
+#include <range/v3/algorithm/any_of.hpp>
+#include <range/v3/algorithm/find_if.hpp>
+#include <range/v3/algorithm/remove_if.hpp>
+
+static constexpr auto LOG_TAG = "NotificationManager";
+
+const NotificationManager::NotificationHandle&
+NotificationManager::passNotificationToAI(Notification notification) {
+    ALOG_TRACE(LOG_TAG) << "passNotificationToAI";
+    const auto at = ranges::find_if(mNotifications, [&](const NotificationHandle& h) {
+        return notification.priority > h.notification.priority;
+    });
+    const auto& result = *mNotifications.emplace(at, NotificationHandle { .notification = std::move(notification) });
+
+    if (result.notification.pin) {
+        // wake up suitable worker based on pin.
+        for (const auto& worker : mWorkers) {
+            if (worker.pins.contains(*result.notification.pin)) {
+                worker.wakeUp.supplyValue();
+                return result;
+            }
+        }
+    }
+
+    // wake up first idle worker.
+    for (const auto& worker : mWorkers) {
+        if (!worker.wakeUp.hasValue()) {
+            worker.wakeUp.supplyValue();
+            return result;
+        }
+    }
+
+    return result;
+
+}
+
+void NotificationManager::removeNotifications(const AString& substring) {
+    ALOG_TRACE(LOG_TAG) << "removeNotifications: " << substring;
+    mNotifications.erase(ranges::remove_if(mNotifications, [&](const NotificationHandle& h) {
+        return h.notification.message.contains(substring);
+    }), mNotifications.end());
+}
+
+AOptional<NotificationManager::NotificationHandle>
+NotificationManager::nextNotification(ASet<AString>& pins) {
+    auto take = [&](std::deque<NotificationHandle>::const_iterator it) {
+        auto notification = std::move(*it);
+        mNotifications.erase(it);
+        if (notification.notification.pin) {
+            pins << *notification.notification.pin;
+        }
+        return notification;
+    };
+    for (auto it = mNotifications.begin(); it != mNotifications.end(); ++it) {
+        if (!it->notification.pin) {
+            return take(it);
+        }
+        if (pins.contains(*it->notification.pin)) {
+            return take(it);
+        }
+        if (ranges::any_of(mWorkers, [&](const Worker& worker) {
+            return worker.pins.contains(*it->notification.pin);
+        })) {
+            // this notification is pinned to other worker, skip.
+            continue;
+        }
+        return take(it);
+    }
+    return std::nullopt;
+}