diff --git a/src/AppBase.cpp b/src/AppBase.cpp index ec48dea..c535dc2 100644 --- a/src/AppBase.cpp +++ b/src/AppBase.cpp @@ -4,129 +4,34 @@ #include "AppBase.h" +#include "IOpenAIChat.h" + #include #include -#include -#include #include #include #include -#include +#include #include "AUI/Logging/ALogger.h" -#include "AUI/Thread/AEventLoop.h" -#include "AUI/Thread/AThreadPool.h" #include "AUI/Util/kAUI.h" -#include "IOpenAIChat.h" -#include "OpenAIChatImpl.h" #include "config.h" #include "MetricsBreadcumbs.h" #include "WebSearch.h" #include "AUI/IO/AFileInputStream.h" #include "tools/ask.h" -#include "util/cosine_similarity.h" #include "util/diary_save_entries.h" #include "util/important_things_to_remember.h" -#include - -static std::default_random_engine re(std::time(nullptr)); +#include using namespace std::chrono_literals; static constexpr auto LOG_TAG = "App"; static const auto WORKING_MEMORY_PATH = "working_memory.md"; +extern std::default_random_engine gRandomEngine; -AFuture> contextEmbedding(IOpenAIChat& openAI, ranges::range auto && rng) { - ALOG_TRACE(LOG_TAG) << "contextEmbedding"; - AString basePrompt; - AUI_ASSERT(!ranges::empty(rng)); - for (const IOpenAIChat::Message& message: rng) { - if (!message.reasoning.empty()) { - basePrompt += message.reasoning; - basePrompt += "\n\n"; - } - if (!message.reasoning_content.empty()) { - basePrompt += message.reasoning_content; - basePrompt += "\n\n"; - } - basePrompt += message.content; - basePrompt += "\n\n---\n\n"; - } - co_return co_await openAI.embedding({ .config = config().embedding }, basePrompt); -} - -[[nodiscard]] -static AFuture<> processRandomlyGoSleep(bool& wakeUp) { - if (config().randomlyGoSleep) { - if (std::uniform_real_distribution(0.0, 1.0)(re) < 0.01) { - // 1. randomly go afk is humane - // 2. reduce resource usage: - // - less conversations would be made - // - in case of group chats and telegram channels, messages would be processed in batches - const auto duration = std::chrono::minutes(std::uniform_int_distribution(15, 120)(re)); - ALogger::info(LOG_TAG) << "Going to sleep for " << std::chrono::duration_cast(duration).count() << " minutes"; - wakeUp = false; - for (int i = 0; i < std::chrono::duration_cast(duration).count(); ++i) { - // костыль ну да сойдёт - if (wakeUp) { - ALogger::info(LOG_TAG) << "Early wake up"; - break; - } - co_await AThread::asyncSleep(1s); - } - } - } -} - -[[nodiscard]] -static AFuture<> processShortcutOpen(AppBase::Notification& notification) { - if (notification.actions.handlers().size() == 1) { - const auto& action = notification.actions.handlers().begin()->second; - if (action.name == "open" && action.parameters.properties.size() == 0) { - // shortcut/optimization: if the notification gives the only option to open it, there's no - // need to ask LLM whether it wants to open the notification because it does it - // in 100% cases. - // Also, this greatly fits in the current architecture, because we can't change notification - // text at runtime, BUT we can provide more recent data by giving the notification code - // control by calling "open()". - notification.message = co_await action.handler({ - .tools = notification.actions, - .args = AJson {}, - .allToolCalls = {}, - }); - } - } -} - -[[nodiscard]] -static bool processIgnoreChance(IOpenAIChat::Session& temporaryContext, bool& canIgnore, const IOpenAIChat::Message& lastLLMResponse) { - if (std::uniform_real_distribution(0.f, 1.f)(re) < config().suggestIgnoreChance) { // attempt to make LLM lazy and ignore message :) - // if (std::exchange(canIgnore, false)) - { // avoid subsequent knockbacks - for (const auto& tc : lastLLMResponse.tool_calls) { - if (tc.function.name == "wait" || tc.function.name == "pause") { - return false; - } - temporaryContext << IOpenAIChat::Message{ - .role = IOpenAIChat::Message::Role::TOOL, - .content = "Error: do you really want to continue? Think again; repeat `{}` to continue or call wait() to finish."_format(AStringView(tc.function.name)), - .tool_call_id = tc.id, - }; - } - ALogger::info(LOG_TAG) << "Begging LLM to be lazy (ignore message)"; - return true; - } - } - return false; -} - -AppBase::~AppBase() { - if (mAliveToken) { - *mAliveToken = false; - } -} AppBase::AppBase(Init init): mInit(std::move(init)), mDiary({ .diaryDir = mInit.workingDir / "diary", @@ -151,278 +56,43 @@ AppBase::AppBase(Init init): mInit(std::move(init)), mDiary({ // }); connect(mWakeupTimer->fired, [this] { - if (std::uniform_real_distribution(0.0, 1.0)(re) < 0.5) { + if (std::uniform_real_distribution(0.0, 1.0)(gRandomEngine) < 0.5) { return; } actProactively(); }); mWakeupTimer->start(); - auto alive = mAliveToken; - getThread()->enqueue([this, alive] { - if (!*alive) return; - mAsync << [](AppBase& self) -> AFuture<> { - // co_await self.mDiary.sleepingConsolidation(); - - co_await self.onBeforeMainLoop(); - for (;;) { - self.onOffline(); - if (self.mSystemPromptSuffix.empty()) { - // this thing emulates "middle" memory of human - tasks, promises and other stuff - // in timespan 1-3d. - self.mSystemPromptSuffix = co_await self.onCleanContext(); - } - #ifndef AUI_TESTS_MODULE - co_await processRandomlyGoSleep(self.mWakeup); - #endif - AUI_ASSERT(AThread::current() == self.getThread()); - if (self.mNotifications.empty()) { - co_await self.mNotificationsSignal; - } - AUI_ASSERT(AThread::current() == self.getThread()); - self.mNotificationsSignal = AFuture<>(); // reset - if (self.mNotifications.empty()) { - continue; - } - auto notification = std::move(self.mNotifications.front()); - self.mNotifications.pop_front(); - self.mAskCalledThisTurn = false; - notification.message += "\nCurrent time: {} UTC"_format(std::chrono::system_clock::now()); - notification.onStartedProcessing.supplyValue(); - AUI_DEFER { notification.onProcessed.supplyValue(); }; - try { - bool canIgnore = true; - co_await processShortcutOpen(notification); - - ALOG_DEBUG(LOG_TAG) << "Processing notification: " << notification.message; - - self.mTemporaryContext << IOpenAIChat::Message{ - .role = IOpenAIChat::Message::Role::USER, - .content = std::move(notification.message), - }; - - // naxyi was here. - // the reasons why I have moved it below diary lookup: - // 1. Each lookup adds ~1s delay. So each time LLM uses send_telegram_message, there is a diary lookup. - // 2. Once again send_telegram_message. Instead of one big message, LLM is encouraged to send multiple small - // messages instead (in the chatting culture the latter is more natural). When we insert occasional - // diary entries between LLMs send_telegram_message calls, it simply loses its focus and starts to spam - // with messages filled with random cues from the diary. - // - // This feels like your participant has ADHD, and they can't finish their thought; instead they remember - // random fact from their sick brain and start yelling "DID YOU KNOW U SHOULD SHIT STANDING UPRIGHT" - // while didn't finish their explanation on why c++ is better than rust. - bool pauseFlag = false; - naxyi_populate_ctx: - if (!self.mDiary.list().empty()) { - AString diary; - - // performs scan on diary based on entire context. - // this will find common cues which are related to current conversation. - if (config().diaryInjectionMaxLength > 0) { - auto currentContext = co_await contextEmbedding(*self.openAI(), self.mTemporaryContext | ranges::view::take_last(3)); - auto relatednesses = co_await self.mDiary.query(currentContext, {.confidenceFactor = 0.f}); - - for (const auto& i : relatednesses) { - const auto&[entryIt, relatedness] = i; - if (relatedness < self.mRelevanceThreshold) { - if (diary.empty()) { - // relax threshold for future queries. - self.mRelevanceThreshold = glm::mix(0.5f, float(relatedness), 0.9f); - } - break; - } - if (diary.length() >= config().diaryInjectionMaxLength) { - // set the minimum constraint for the future queries - self.mRelevanceThreshold = relatedness; - break; - } - diary += self.takeDiaryEntry(i); - } - } - - if (!diary.empty()) { - diary += self.mTemporaryContext.last().content; - self.mTemporaryContext.last().content = std::move(diary); - } - } - - naxyi_preserve_ctx: - self.updateTools(notification.actions); - if (!self.mAskCalledThisTurn) { - // remind LLM to call #ask before responding. - // Injected as a system-level checkpoint so LLM sees it right before generating its next action. - if (config().remindUseAsk) { - self.mTemporaryContext.last().content += - "\n[system] Have you called #ask yet this turn? " - "If the message involves personal topics, past events, questions, or people you know — " - "call #ask BEFORE send_telegram_message."; - } - } - auto escape = [&](OpenAITools::Ctx ctx) -> AFuture { - pauseFlag = true; - if (self.mActingProactively) { - // at the end of "actProactively", let's try to encourage LLM to write someone, still. - // if LLM's haven't written to anyone at this point, this notification will guide the LLM - // that dismissive action is not acceptable and LLM will try to revisit some older dialog - // despite no cue. - // if LLM actually have written to someone at this point, LLM will initiate a dialog with - // one more person. - self.passNotificationToAI("You should write someone else and be more proactive.", {}, true); - } - co_return "Success"; - }; - notification.actions.insert({ - .name = "pause", - .description = "Pauses the conversation", - .handler = escape, - }); - notification.actions.insert({ - .name = "wait", - .description = "Wait until further notifications", - .handler = escape, - }); - IOpenAIChat::Response botAnswer = co_await [&]() -> AFuture { - MetricsBreadcumbs::Point metric(self.metricBreadcumbs(), "function", "notification processing loop"); - auto response = self.openAI()->chatStreaming( { - .systemPrompt = self.getSystemPrompt(), - .tools = notification.actions.asJson(), - }, self.mTemporaryContext); - connect(response->response.changed, self, [&self](IOpenAIChat::Response response) { - self.onResponseAssembling(std::move(response)); - }); - co_await response->completed; - co_return std::move(*response->response); - }(); - AUI_ASSERT(AThread::current() == self.getThread()); - - if (botAnswer.choices.empty() || botAnswer.choices.at(0).message.tool_calls.empty()) { - // no tool calls. - // each LLMs turn should end with "wait" or "pause" - ALogger::warn(LOG_TAG) << "LLM didn't perform any action."; - if (!botAnswer.choices.empty()) { - // guiderails to make LLM tool-centric. - const auto& content = botAnswer.choices.at(0).message.content; - if (content.contains("#send_telegram_message")) { - // qwen3.5 bug: misused examples - self.mTemporaryContext << IOpenAIChat::Message{ - .role = IOpenAIChat::Message::Role::USER, - .content = "Nice thoughts! However you should be tool-centric. Make sure you " - "made tool calls. The message you provided is not visible to anyone but you.", - }; - goto naxyi_preserve_ctx; - } - if (content.contains("")) { - // gemma4 bug: does not perform tool calls, instead, replies with the following content - // - // Ой, и что же ты там читаешь? Надеюсь, только самое милое! 😼✨ - // - - self.mTemporaryContext << IOpenAIChat::Message{ - .role = IOpenAIChat::Message::Role::USER, - .content = "Nice thoughts! However you should be tool-centric. Make sure you " - "made tool calls. The message you provided is not visible to anyone but you. Call " - "#wait if you are unsure.", - }; - goto naxyi_preserve_ctx; - } - } - // punish llm for not performing tool calls. - self.mTemporaryContext << IOpenAIChat::Message{ - .role = IOpenAIChat::Message::Role::USER, - .content = "Nice thoughts! However you should be tool-centric. Make sure you " - "made tool calls. The message you provided is not visible to anyone but you. Call #wait if " - "you are unsure.", - }; - goto naxyi_preserve_ctx; - } - - if (processIgnoreChance(self.mTemporaryContext, canIgnore, botAnswer.choices.at(0).message)) { - goto naxyi_preserve_ctx; - } - - { - auto toolCalls = co_await notification.actions.handleToolCalls(botAnswer.choices.at(0).message.tool_calls, self.metricBreadcumbs()); - if (ranges::any_of(toolCalls, [](const IOpenAIChat::Message& msg) { return msg.content.contains(IOpenAIChat::EMBEDDING_TAG); })) { - // Indicates a low quality tool call. - // - // This tag is used as an exception condition within a tool handler, and handled by AppBase. - // When caught, LLM's tool call appends to the user's last message, and user's last message will - // be sent again. - // - // This allows the feedback workflow: when a low quality message was passed to - // send_telegram_message, it can throw EMBEDDING_TAG to rollback before LLM's - // #send_telegram_message and slightly adjust LLM's following action. This differs from the - // standard AException workflow which is used for technical errors (such as you were banned, or - // no internet connection) whose are meaningful to LLM and it can adopt to. - - if (botAnswer.usage.prompt_tokens > config().diaryTokenCountTrigger) { - // we are stuck; ignore the event - ALogger::warn("AppBase") << "LLM can't find proper response to the notification; " - "context is overflown. Ignoring event and dumping context"; - co_await self.diaryDumpMessages(); - continue; - } - goto naxyi_preserve_ctx; - } - self.mTemporaryContext << botAnswer.choices.at(0).message; - self.mTemporaryContext << std::move(toolCalls); - ALOG_DEBUG(LOG_TAG) << "Tool call response: " << self.mTemporaryContext.last().content; - AUI_ASSERT(AThread::current() == self.getThread()); - } + auto fetchConfig = [this] { + mWorkerCount = config().workerCount; + }; + fetchConfig(); + connect(gConfigUpdated, fetchConfig); - if (pauseFlag) { - finish: - if (botAnswer.usage.total_tokens >= config().diaryTokenCountTrigger) { - co_await self.diaryDumpMessages(); - } - continue; - } - if (!notification.actions.handlers().empty()) { - self.mTemporaryContext.last().content += "\nWhat's your next action? Use a `tool` to act. Use #ask to consult with your knowledge database. The following tools available: " + AStringVector(notification.actions.handlers().keyVector()).join(", "); - } - if (ranges::any_of(botAnswer.choices.at(0).message.tool_calls, [](const IOpenAIChat::Message::ToolCall& t){ return t.function.name == "send_telegram_message"; })) { - // if LLM sent a message without ever calling #ask this turn, - // inject a reminder into the next turn's context. - if (!self.mAskCalledThisTurn && config().remindUseAsk) { - self.mTemporaryContext.last().content += - "\n[system] Note: you sent a message without consulting #ask this turn. " - "Next time, call #ask before send_telegram_message to enrich your response " - "with memories and context."; - } - goto naxyi_preserve_ctx; - } else { - goto naxyi_populate_ctx; - } - } catch (const AException& e) { - ALogger::err(LOG_TAG) << "Failed to process notification: \"" << notification.message << "\"" << e; - if (e.getMessage().lowercase().contains("json")) { - // If there's a JSON error, it means we have irreversibly damaged context. Best way to solve this - // is to drop the temporary context entirery. - ALogger::warn("AppBase") << "Context is damaged. Dropping context"; - self.mTemporaryContext.clear(); - } - } + connect(mWorkerCount, [this] { + if (mWorkerCount > mWorkers.size()) { + for (size_t i = mWorkerCount - mWorkers.size(); i > 0; --i) { + mWorkers << _new(mWorkers.size(), *this); } - co_return; - }(*this); + return; + } + if (mWorkerCount < mWorkers.size()) { + mWorkers.erase(mWorkers.end() - mWorkerCount, mWorkers.end()); + } }); } -const AppBase::Notification& AppBase::passNotificationToAI(AString notification, OpenAITools actions, bool first) { - ALOG_TRACE(LOG_TAG) << "passNotificationToAI"; - const auto& result = *mNotifications.emplace(first ? mNotifications.begin() : mNotifications.end(), std::move(notification), std::move(actions)); - mNotificationsSignal.supplyValue(); - return result; -} - -AFuture<> AppBase::diaryDumpMessages() { +AFuture<> AppBase::diaryDumpMessages(IOpenAIChat::Session& temporaryContext) { + std::unique_lock lock(mWorkingMemoryLock, std::defer_lock); + while (!lock.try_lock()) { + co_await AThread::asyncSleep(1s); + } MetricsBreadcumbs::Point metric(metricBreadcumbs(), "function", "diaryDumpMessages"); ALOG_TRACE(LOG_TAG) << "diaryDumpMessages"; // mDiary.reload(); // will find plagiarism against all entries. // commented out: exclude plagiarism checks for // included entries AUI_DEFER { mDiary.reload(); }; - if (mTemporaryContext.empty()) { + if (temporaryContext.empty()) { co_return; } AString previousWorkingMemory; @@ -431,9 +101,9 @@ AFuture<> AppBase::diaryDumpMessages() { buf << AFileInputStream(mInit.workingDir / WORKING_MEMORY_PATH); previousWorkingMemory = AStringView(buf.data(), buf.size()); } - auto importantThingsToRemember = util::importantThingsToRemember(*this, *openAI(), mTemporaryContext, previousWorkingMemory); + auto importantThingsToRemember = util::importantThingsToRemember(*this, *openAI(), temporaryContext, previousWorkingMemory); - co_await util::diarySaveEntries(mDiary, mTemporaryContext, { + co_await util::diarySaveEntries(mDiary, temporaryContext, { .systemPrompt = getSystemPrompt(), // no tools should be involved. }); @@ -442,7 +112,7 @@ AFuture<> AppBase::diaryDumpMessages() { auto workingMemoryMd = co_await importantThingsToRemember; AFileOutputStream(mInit.workingDir / WORKING_MEMORY_PATH) << workingMemoryMd; } - mTemporaryContext.clear(); + temporaryContext.clear(); mSystemPromptSuffix.clear(); } @@ -450,7 +120,7 @@ void AppBase::actProactively() { ALOG_TRACE(LOG_TAG) << "actProactively"; AString prompt = "\n"; if (!mDiary.list().empty()) { - auto idx = re() % mDiary.list().size(); + auto idx = gRandomEngine() % mDiary.list().size(); auto entry = mDiary.list().begin(); while (idx--) { entry++; @@ -468,7 +138,10 @@ It's time to reflect on your thoughts! can open one chat at a time - choose wisely!\n" Act proactively! )"; - const auto& notification = passNotificationToAI(std::move(prompt)); + const auto& notification = notificationManager().passNotificationToAI({ + .message = std::move(prompt), + .pin = "", + }); struct State { AOptional metric; }; @@ -483,11 +156,11 @@ Act proactively! }); } -AFuture AppBase::onCleanContext() { +AString AppBase::onCleanContext() const { if ((mInit.workingDir / WORKING_MEMORY_PATH).isRegularFileExists()) { AByteBuffer workingMemory; workingMemory << AFileInputStream(mInit.workingDir / WORKING_MEMORY_PATH); - co_return R"( + return R"( {} @@ -506,30 +179,27 @@ send_telegram_message("text":"мррр~") )"_format(AStringView(workingMemory.data(), workingMemory.size())); } - co_return ""; + return ""; } -void AppBase::updateTools(OpenAITools& actions) { +void AppBase::updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext) { ALOG_TRACE(LOG_TAG) << "updateTools"; - actions.insert(tools::ask([this] { + actions.insert(tools::ask([&temporaryContext] { AString out; - for (const auto& msg : mTemporaryContext | ranges::view::take_last(2)) { + for (const auto& msg : temporaryContext | ranges::view::take_last(2)) { out += msg.content; out += "\n"; } return out; }, openAI(), mDiary)); - actions.onAfterToolCall = [this](const AString& toolName) { + actions.onAfterToolCall << [this](const AString& toolName) { if (toolName == "wait") { return; } if (toolName == "pause") { return; } - if (toolName == "ask") { - mAskCalledThisTurn = true; - } auto labels = metricBreadcumbs()->value(); emit toolCallFired(AppBase::ToolCallEvent{ .toolName = toolName, @@ -542,39 +212,29 @@ void AppBase::updateTools(OpenAITools& actions) { } -void AppBase::removeNotifications(const AString& substring) { - ALOG_TRACE(LOG_TAG) << "removeNotifications: " << substring; - mNotifications.erase(ranges::remove_if(mNotifications, [&](const Notification& n) { - return n.message.contains(substring); - }), mNotifications.end()); -} - -AString AppBase::takeDiaryEntry(const Diary::EntryExAndRelatedness& i) { - ALOG_TRACE(LOG_TAG) << "takeDiaryEntry: " << i.entry->id; - if (ranges::any_of(mTemporaryContext, [&](const IOpenAIChat::Message& m) { - return m.content.contains(i.entry->freeformBody); - })) { - // if mTemporaryContext already contains this diary entry verbatim - we don't need to reinclude it - it makes - // no sense to consume tokens for the same thing. - // - // the copypasted diary entry would not receive score. - return {}; +void AppBase::wakeUpIfSleeping() { + for (const auto& worker : mWorkers) { + worker->wakeUpIfSleeping(); } - - i.entry->metadata.score += (i.relatedness - 0.5f) * 2.f; - i.entry->incrementUsageCount(); - ALogger::info("AppBase") << "Loaded into context: " << i.entry->id << ".md relatedness=" << i.relatedness << "\n" << i.entry->freeformBody; - auto formattedTag = "{} additional_context just_for_reasoning no_plagiarism no_copy"_format("your_diary_page"); - AString result = "<{}>\n{}\n\n"_format(formattedTag, i.entry->freeformBody, formattedTag); - mDiary.unload(i.entry); - return result; } -AString AppBase::getSystemPrompt() const +AString AppBase::getSystemPrompt() { + if (mSystemPromptSuffix.empty()) { + // this thing emulates "middle" memory of human - tasks, promises and other stuff + // in timespan 1-3d. + // reset in diaryDumpMessages. + mSystemPromptSuffix = onCleanContext(); + } auto prompt = "{}\n\n{}\n\n\n{}\n\n{}"_format(prompts().system, prompts().characterBase, prompts().characterAppearance, mSystemPromptSuffix); if (config().lockdown != Config::LockdownMode::NONE) { prompt += "\nYou are currently under lockdown mode. This means you can access to subset of chats only."; } return prompt; } + +AFuture<> AppBase::diaryDumpMessages() { + for (const auto& worker : mWorkers) { + co_await worker->diaryDumpMessages(); + } +} diff --git a/src/AppBase.h b/src/AppBase.h index baec1bf..1c63dc7 100644 --- a/src/AppBase.h +++ b/src/AppBase.h @@ -1,5 +1,5 @@ -#include #pragma once +#include #include "AUI/Common/AObject.h" #include "AUI/Common/ATimer.h" #include "AUI/Thread/AAsyncHolder.h" @@ -7,7 +7,9 @@ #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 { _ 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 mLastOpenedChatLastMessageTime; - - virtual AFuture 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 _& 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 mLastOpenedChatLastMessageTime; - [[nodiscard]] - const _& openAI() const noexcept { - return mInit.openAI; - } private: _ mMetricBreadcumbs = _new(); const Init mInit; - std::deque mNotifications; AFuture<> mNotificationsSignal; _ mWakeupTimer; - // OpenAITools mTools; + NotificationManager mNotificationManager; AString mSystemPromptSuffix; bool mWakeup = false; @@ -161,6 +105,9 @@ class AppBase : public AObject { */ bool mAskCalledThisTurn = false; + AVector> mWorkers; + AProperty mWorkerCount; + ASpinlockMutex mWorkingMemoryLock; + Diary mDiary; - std::shared_ptr mAliveToken = std::make_shared(true); }; diff --git a/src/Diary.h b/src/Diary.h index 55ef703..70275da 100644 --- a/src/Diary.h +++ b/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> query(const std::valarray& query, QueryOpts opts); + AFuture> query(const std::valarray& 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. */ diff --git a/src/ImageGenerator.cpp b/src/ImageGenerator.cpp index 2a3b12b..77bacdd 100644 --- a/src/ImageGenerator.cpp +++ b/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::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::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 = "" }; diff --git a/src/NotificationManager.cpp b/src/NotificationManager.cpp new file mode 100644 index 0000000..2996b01 --- /dev/null +++ b/src/NotificationManager.cpp @@ -0,0 +1,76 @@ +// +// Created by alex2772 on 7/14/26. +// + +#include "NotificationManager.h" + +#include +#include +#include + +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::nextNotification(ASet& pins) { + auto take = [&](std::deque::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; +} diff --git a/src/NotificationManager.h b/src/NotificationManager.h new file mode 100644 index 0000000..5c246f4 --- /dev/null +++ b/src/NotificationManager.h @@ -0,0 +1,115 @@ +#pragma once +#include "OpenAITools.h" +#include "AUI/Util/kAUI.h" +#include "AUI/Thread/AFuture.h" + +#include +#include +#include + +class NotificationManager { +public: + struct Notification { + /** + * @brief Notification message in natural language passed to LLM. + * @details + * Example: You have received a message from User. Would you like to process it? + */ + AString message; + + /** + * @brief Related tools available to the LLM when processing this notification. + * @details + * Example: "open" tool to open the chat the notification came from. + */ + OpenAITools actions; + + /** + * @brief Priority of the notification. Higher priority notifications will be processed earlier. + */ + int priority = 0; + + /** + * @brief Optional freeform pin token that poisons the worker that will process the notification. + * @details + * Example pin string: "". + * + * The worker pinning mechanism was introduced to pin a chat to specific worker, so the same chat is processed + * by the same worker, which has more related context. + */ + AOptional pin; + }; + + struct NotificationHandle { + Notification notification; + + /** + * @brief Resolves when the notification was passed to the worker. + */ + AFuture<> onStartedProcessing; + + /** + * @brief Resolved by the worker when the notification pass completely processed. + */ + AFuture<> onProcessed; + }; + + /** + * @brief Registers a worker and runs it in a loop. + * @param workerPins a set of string that identifies chats and kinds of events this worker has processed. This + * allows to route notifications of the same chat to the same worker. + * @param worker worker callback. accepts notifications. returns false to break. + */ + template + requires requires(WorkerCallback&& workerCallback, Notification h) { + { workerCallback(std::move(h)) } -> aui::same_as>; + } + AFuture<> run(ASet& workerPins, WorkerCallback worker) { + auto workerRegistrationToken = mWorkers.insert(mWorkers.end(), Worker { workerPins }); + AUI_DEFER { mWorkers.erase(workerRegistrationToken); }; + while (true) { + auto handle = nextNotification(workerPins); + if (!handle) { + co_await workerRegistrationToken->wakeUp; + workerRegistrationToken->wakeUp = AFuture<>(); // reset + continue; + } + handle->notification.message += "\nCurrent time: {} UTC"_format(std::chrono::system_clock::now()); + handle->onStartedProcessing.supplyValue(); + AUI_DEFER { handle->onProcessed.supplyValue(); }; + if (!co_await worker(std::move(handle->notification))) { + break; + } + } + } + + /** + * @brief Passes an event to the AI to process + * Think of it as your phone's notifications: you receive a notification, read it and (maybe) react to it. + */ + const NotificationHandle& passNotificationToAI(Notification notification); + + /** + * @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 the queue. + */ + void removeNotifications(const AString& substring); + +private: + std::deque mNotifications; + struct Worker { + const ASet& pins; + AFuture<> wakeUp; + }; + std::list mWorkers; + + /** + * @brief Finds next notification to process. + * @param pins Worker's pins. + * @return + */ + AOptional nextNotification(ASet& pins); + +}; \ No newline at end of file diff --git a/src/OpenAITools.cpp b/src/OpenAITools.cpp index d32c157..cd03aa4 100644 --- a/src/OpenAITools.cpp +++ b/src/OpenAITools.cpp @@ -41,8 +41,8 @@ static AString removeControlCharacters(AString input) { } AFuture OpenAITools::handleToolCalls(const AVector& toolCalls, - const _& metricsBreadCumbs) { - ALOG_TRACE("OpenAITools") << "handleToolCalls"; + const _& metricsBreadCumbs, const IOpenAIChat::Session& temporaryContext, ALogger& logger) { + logger.trace("OpenAITools") << "handleToolCalls"; IOpenAIChat::Session result; for (const auto& toolCall : toolCalls) { result << IOpenAIChat::Message{ @@ -55,18 +55,20 @@ AFuture OpenAITools::handleToolCalls(const AVectorsecond.handler({ + .logger = logger, .tools = *this, .args = AJson::fromString(toolCall.function.arguments), + .temporaryContext = temporaryContext, .allToolCalls = toolCalls, }); - if (onAfterToolCall) { - onAfterToolCall(toolCall.function.name); + for (const auto& i : onAfterToolCall) { + i(toolCall.function.name); } co_return std::move(handlerResult); } co_return "tool \"" + toolCall.function.name + "\" is not currently available. Please use another tool instead."; } catch (const AException& e) { - ALogger::err("OpenAITools") << "error while executing \"{}\" tool: "_format(toolCall.function.name) << e; + logger.err("OpenAITools") << "error while executing \"{}\" tool: "_format(toolCall.function.name) << e; co_return "error while executing \"{}\" tool: {}"_format(toolCall.function.name, e.getMessage()); } }()), diff --git a/src/OpenAITools.h b/src/OpenAITools.h index 45b87a4..864fa91 100644 --- a/src/OpenAITools.h +++ b/src/OpenAITools.h @@ -7,8 +7,10 @@ struct OpenAITools { struct Ctx { + ALogger& logger = ALogger::global(); OpenAITools& tools; AJson args; + const IOpenAIChat::Session& temporaryContext; const AVector& allToolCalls; }; using Handler = std::function(Ctx ctx)>; @@ -37,9 +39,9 @@ struct OpenAITools { * @brief Optional hook fired after each tool call handler completes successfully. * Not called if the handler throws. Set by AppBase::updateTools to emit AppBase::toolCallFired. */ - std::function onAfterToolCall; + AVector> onAfterToolCall; - AFuture handleToolCalls(const AVector& toolCalls, const _& metricsBreadCumbs = nullptr); + AFuture handleToolCalls(const AVector& toolCalls, const _& metricsBreadCumbs = nullptr, const IOpenAIChat::Session& temporaryContext = {}, ALogger& logger = ALogger::global()); AJson asJson() const; diff --git a/src/Worker.cpp b/src/Worker.cpp new file mode 100644 index 0000000..83b38b5 --- /dev/null +++ b/src/Worker.cpp @@ -0,0 +1,413 @@ +// +// Created by alex2772 on 7/14/26. +// + +#include "Worker.h" + +#include + +#include "AppBase.h" +#include "NotificationManager.h" +#include "AppBase.h" +#include "IOpenAIChat.h" + +#include + +static constexpr auto LOG_TAG = "Worker"; + +using namespace std::chrono_literals; + +extern std::default_random_engine gRandomEngine; + + +AFuture> contextEmbedding(ALogger& logger, IOpenAIChat& openAI, ranges::range auto&& rng) { + logger.trace(LOG_TAG) << "contextEmbedding"; + AString basePrompt; + AUI_ASSERT(!ranges::empty(rng)); + for (const IOpenAIChat::Message& message : rng) { + if (!message.reasoning.empty()) { + basePrompt += message.reasoning; + basePrompt += "\n\n"; + } + if (!message.reasoning_content.empty()) { + basePrompt += message.reasoning_content; + basePrompt += "\n\n"; + } + basePrompt += message.content; + basePrompt += "\n\n---\n\n"; + } + co_return co_await openAI.embedding({ .config = config().embedding }, basePrompt); +} + +[[nodiscard]] +static AFuture<> processRandomlyGoSleep(ALogger& logger, bool& wakeUp) { + if (config().randomlyGoSleep) { + if (std::uniform_real_distribution(0.0, 1.0)(gRandomEngine) < 0.01) { + // 1. randomly go afk is humane + // 2. reduce resource usage: + // - less conversations would be made + // - in case of group chats and telegram channels, messages would be processed in batches + const auto duration = std::chrono::minutes(std::uniform_int_distribution(15, 120)(gRandomEngine)); + logger.info(LOG_TAG) + << "Going to sleep for " << std::chrono::duration_cast(duration).count() << " minutes"; + wakeUp = false; + for (int i = 0; i < std::chrono::duration_cast(duration).count(); ++i) { + // костыль ну да сойдёт + if (wakeUp) { + logger.info(LOG_TAG) << "Early wake up"; + break; + } + co_await AThread::asyncSleep(1s); + } + } + } +} + +[[nodiscard]] +static AFuture<> processShortcutOpen(NotificationManager::Notification& notification, const IOpenAIChat::Session& temporaryContext) { + if (notification.actions.handlers().size() == 1) { + const auto& action = notification.actions.handlers().begin()->second; + if (action.name == "open" && action.parameters.properties.size() == 0) { + // shortcut/optimization: if the notification gives the only option to open it, there's no + // need to ask LLM whether it wants to open the notification because it does it + // in 100% cases. + // Also, this greatly fits in the current architecture, because we can't change notification + // text at runtime, BUT we can provide more recent data by giving the notification code + // control by calling "open()". + notification.message = co_await action.handler({ + .tools = notification.actions, + .args = AJson {}, + .temporaryContext = temporaryContext, + .allToolCalls = {}, + }); + } + } +} + +[[nodiscard]] +static bool +processIgnoreChance(ALogger& logger, IOpenAIChat::Session& temporaryContext, bool& canIgnore, const IOpenAIChat::Message& lastLLMResponse) { +#ifdef AUI_TESTS_MODULE + if (std::uniform_real_distribution(0.f, 1.f)(gRandomEngine) < config().suggestIgnoreChance) { // attempt to make + // LLM lazy and + // ignore message :) + // if (std::exchange(canIgnore, false)) + { // avoid subsequent knockbacks + for (const auto& tc : lastLLMResponse.tool_calls) { + if (tc.function.name == "wait" || tc.function.name == "pause") { + return false; + } + temporaryContext << IOpenAIChat::Message { + .role = IOpenAIChat::Message::Role::TOOL, + .content = + "Error: do you really want to continue? Think again; repeat `{}` to continue or call wait() to finish."_format( + AStringView(tc.function.name)), + .tool_call_id = tc.id, + }; + } + logger.info(LOG_TAG) << "Begging LLM to be lazy (ignore message)"; + return true; + } + } +#endif + return false; +} + +AFuture<> Worker::handleNotification(std::shared_ptr alive, NotificationManager::Notification notification) { +#ifndef AUI_TESTS_MODULE + co_await processRandomlyGoSleep(mLogger, mWakeUp); +#endif + AUI_ASSERT(AThread::current() == getThread()); + AUI_DEFER { mApp.onOffline(); }; + AUI_ASSERT(*alive); + mAskCalledThisTurn = false; + try { + bool canIgnore = true; + co_await processShortcutOpen(notification, mTemporaryContext); + + mLogger.info(LOG_TAG) << "Processing notification: " << notification.message; + + mTemporaryContext << IOpenAIChat::Message { + .role = IOpenAIChat::Message::Role::USER, + .content = std::move(notification.message), + }; + + // naxyi was here. + // the reasons why I have moved it below diary lookup: + // 1. Each lookup adds ~1s delay. So each time LLM uses send_telegram_message, there is a diary + // lookup. + // 2. Once again send_telegram_message. Instead of one big message, LLM is encouraged to send + // multiple small + // messages instead (in the chatting culture the latter is more natural). When we insert + // occasional diary entries between LLMs send_telegram_message calls, it simply loses its + // focus and starts to spam with messages filled with random cues from the diary. + // + // This feels like your participant has ADHD, and they can't finish their thought; instead + // they remember random fact from their sick brain and start yelling "DID YOU KNOW U SHOULD + // SHIT STANDING UPRIGHT" while didn't finish their explanation on why c++ is better than + // rust. + bool pauseFlag = false; + naxyi_populate_ctx: + if (!mApp.diary().list().empty()) { + AString diary; + + // performs scan on diary based on entire context. + // this will find common cues which are related to current conversation. + if (config().diaryInjectionMaxLength > 0) { + auto currentContext = + co_await contextEmbedding(mLogger, *mApp.openAI(), mTemporaryContext | ranges::view::take_last(3)); + auto relatednesses = co_await mApp.diary().query(currentContext, { .confidenceFactor = 0.f }); + + for (const auto& i : relatednesses) { + const auto& [entryIt, relatedness] = i; + if (relatedness < mRelevanceThreshold) { + if (diary.empty()) { + // relax threshold for future queries. + mRelevanceThreshold = glm::mix(0.5f, float(relatedness), 0.9f); + } + break; + } + if (diary.length() >= config().diaryInjectionMaxLength) { + // set the minimum constraint for the future queries + mRelevanceThreshold = relatedness; + break; + } + diary += takeDiaryEntry(i); + } + } + + if (!diary.empty()) { + diary += mTemporaryContext.last().content; + mTemporaryContext.last().content = std::move(diary); + } + } + + naxyi_preserve_ctx: + updateTools(notification.actions); + if (!mAskCalledThisTurn) { + // remind LLM to call #ask before responding. + // Injected as a system-level checkpoint so LLM sees it right before generating its next + // action. + if (config().remindUseAsk) { + mTemporaryContext.last().content += + "\n[system] Have you called #ask yet this turn? " + "If the message involves personal topics, past events, questions, or people you " + "know — " + "call #ask BEFORE send_telegram_message."; + } + } + auto escape = [&](OpenAITools::Ctx ctx) -> AFuture { + pauseFlag = true; + if (mApp.isActingProactively()) { + // at the end of "actProactively", let's try to encourage LLM to write someone, still. + // if LLM's haven't written to anyone at this point, this notification will guide the + // LLM that dismissive action is not acceptable and LLM will try to revisit some older + // dialog despite no cue. if LLM actually have written to someone at this point, LLM + // will initiate a dialog with one more person. + mApp.notificationManager().passNotificationToAI({ + .message = "You should write someone else and be more proactive.", + .pin = "", + }); + } + co_return "Success"; + }; + notification.actions.insert({ + .name = "pause", + .description = "Pauses the conversation", + .handler = escape, + }); + notification.actions.insert({ + .name = "wait", + .description = "Wait until further notifications", + .handler = escape, + }); + IOpenAIChat::Response botAnswer = co_await [&]() -> AFuture { + MetricsBreadcumbs::Point metric(mApp.metricBreadcumbs(), "function", "notification processing loop"); + auto response = mApp.openAI()->chatStreaming( + { + .systemPrompt = mApp.getSystemPrompt(), + .tools = notification.actions.asJson(), + }, + mTemporaryContext); + connect(response->response.changed, mApp, [&](IOpenAIChat::Response response) { + mApp.onResponseAssembling(std::move(response)); + }); + co_await response->completed; + co_return std::move(*response->response); + }(); + AUI_ASSERT(AThread::current() == getThread()); + + if (botAnswer.choices.empty() || botAnswer.choices.at(0).message.tool_calls.empty()) { + // no tool calls. + // each LLMs turn should end with "wait" or "pause" + mLogger.warn(LOG_TAG) << "LLM didn't perform any action."; + if (!botAnswer.choices.empty()) { + // guiderails to make LLM tool-centric. + const auto& content = botAnswer.choices.at(0).message.content; + if (content.contains("#send_telegram_message")) { + // qwen3.5 bug: misused examples + mTemporaryContext << IOpenAIChat::Message { + .role = IOpenAIChat::Message::Role::USER, + .content = + "Nice thoughts! However you should be tool-centric. Make sure you " + "made tool calls. The message you provided is not visible to anyone but " + "you.", + }; + goto naxyi_preserve_ctx; + } + if (content.contains("")) { + // gemma4 bug: does not perform tool calls, instead, replies with the following + // content Ой, и что же ты там читаешь? Надеюсь, только самое милое! + // 😼✨ + // + + mTemporaryContext << IOpenAIChat::Message { + .role = IOpenAIChat::Message::Role::USER, + .content = + "Nice thoughts! However you should be tool-centric. Make sure you " + "made tool calls. The message you provided is not visible to anyone but " + "you. Call " + "#wait if you are unsure.", + }; + goto naxyi_preserve_ctx; + } + } + // punish llm for not performing tool calls. + mTemporaryContext << IOpenAIChat::Message { + .role = IOpenAIChat::Message::Role::USER, + .content = + "Nice thoughts! However you should be tool-centric. Make sure you " + "made tool calls. The message you provided is not visible to anyone but you. Call " + "#wait if " + "you are unsure.", + }; + goto naxyi_preserve_ctx; + } + + if (processIgnoreChance(mLogger, mTemporaryContext, canIgnore, botAnswer.choices.at(0).message)) { + goto naxyi_preserve_ctx; + } + + { + auto toolCalls = co_await notification.actions.handleToolCalls( + botAnswer.choices.at(0).message.tool_calls, mApp.metricBreadcumbs(), mTemporaryContext, mLogger); + if (ranges::any_of(toolCalls, [](const IOpenAIChat::Message& msg) { + return msg.content.contains(IOpenAIChat::EMBEDDING_TAG); + })) { + // Indicates a low quality tool call. + // + // This tag is used as an exception condition within a tool handler, and handled by + // AppBase. When caught, LLM's tool call appends to the user's last message, and user's + // last message will be sent again. + // + // This allows the feedback workflow: when a low quality message was passed to + // send_telegram_message, it can throw EMBEDDING_TAG to rollback before LLM's + // #send_telegram_message and slightly adjust LLM's following action. This differs from + // the standard AException workflow which is used for technical errors (such as you were + // banned, or no internet connection) whose are meaningful to LLM and it can adopt to. + + if (botAnswer.usage.prompt_tokens > config().diaryTokenCountTrigger) { + // we are stuck; ignore the event + mLogger.warn("AppBase") + << "LLM can't find proper response to the notification; " + "context is overflown. Ignoring event and dumping context"; + co_await diaryDumpMessages(); + co_return; + } + goto naxyi_preserve_ctx; + } + mTemporaryContext << botAnswer.choices.at(0).message; + mTemporaryContext << std::move(toolCalls); + mLogger.info(LOG_TAG) << "Tool call response: " << mTemporaryContext.last().content; + AUI_ASSERT(AThread::current() == getThread()); + } + + if (pauseFlag) { + finish: + if (botAnswer.usage.total_tokens >= config().diaryTokenCountTrigger) { + co_await diaryDumpMessages(); + } + co_return; + } + if (!notification.actions.handlers().empty()) { + mTemporaryContext.last().content += + "\nWhat's your next action? Use a `tool` to act. Use #ask to consult with your " + "knowledge database. The following tools available: " + + AStringVector(notification.actions.handlers().keyVector()).join(", "); + } + if (ranges::any_of(botAnswer.choices.at(0).message.tool_calls, [](const IOpenAIChat::Message::ToolCall& t) { + return t.function.name == "send_telegram_message"; + })) { + // if LLM sent a message without ever calling #ask this turn, + // inject a reminder into the next turn's context. + if (!mAskCalledThisTurn && config().remindUseAsk) { + mTemporaryContext.last().content += + "\n[system] Note: you sent a message without consulting #ask this turn. " + "Next time, call #ask before send_telegram_message to enrich your response " + "with memories and context."; + } + goto naxyi_preserve_ctx; + } else { + goto naxyi_populate_ctx; + } + } catch (const AException& e) { + mLogger.err(LOG_TAG) << "Failed to process notification: \"" << notification.message << "\"" << e; + if (e.getMessage().lowercase().contains("json")) { + // If there's a JSON error, it means we have irreversibly damaged context. Best way to solve + // this is to drop the temporary context entirery. + mLogger.warn("AppBase") << "Context is damaged. Dropping context"; + mTemporaryContext.clear(); + } + } +} + +Worker::Worker(size_t name, AppBase& app): mName(name), mApp(app) { + mAliveToken = _new(true); + + getThread()->enqueue([=, alive = mAliveToken] { + if (!*alive) + return; + mCoroutine = mApp.notificationManager().run(mWorkerPins, [=](NotificationManager::Notification notification) -> AFuture { + co_await handleNotification(alive, std::move(notification)); + co_return *alive; + }); + }); +} + +Worker::~Worker() { *mAliveToken = false; } + +AString Worker::takeDiaryEntry(const Diary::EntryExAndRelatedness& i) { + mLogger.trace(LOG_TAG) << "takeDiaryEntry: " << i.entry->id; + if (ranges::any_of(mTemporaryContext, [&](const IOpenAIChat::Message& m) { + return m.content.contains(i.entry->freeformBody); + })) { + // if mTemporaryContext already contains this diary entry verbatim - we don't need to reinclude it - it makes + // no sense to consume tokens for the same thing. + // + // the copypasted diary entry would not receive score. + return {}; + } + + i.entry->metadata.score += (i.relatedness - 0.5f) * 2.f; + i.entry->incrementUsageCount(); + mLogger.info("AppBase") << "Loaded into context: " << i.entry->id << ".md relatedness=" << i.relatedness << "\n" << i.entry->freeformBody; + auto formattedTag = "{} additional_context just_for_reasoning no_plagiarism no_copy"_format("your_diary_page"); + AString result = "<{}>\n{}\n\n"_format(formattedTag, i.entry->freeformBody, formattedTag); + mApp.diary().unload(i.entry); + return result; +} + +void Worker::updateTools(OpenAITools& tools) { + mApp.updateTools(tools, mTemporaryContext); + tools.onAfterToolCall << [this](const AString& toolName) { + if (toolName == "ask") { + mAskCalledThisTurn = true; + } + }; +} + +AFuture<> Worker::diaryDumpMessages() { + mWorkerPins.clear(); + co_await mApp.diaryDumpMessages(mTemporaryContext); +} diff --git a/src/Worker.h b/src/Worker.h new file mode 100644 index 0000000..3770bea --- /dev/null +++ b/src/Worker.h @@ -0,0 +1,48 @@ +#pragma once +#include "AUI/Common/AObject.h" +#include "AUI/Thread/AAsyncHolder.h" +#include "IOpenAIChat.h" +#include "NotificationManager.h" +#include "Diary.h" + +#include + +class AppBase; + +class Worker: public AObject { +public: + Worker(size_t name, AppBase& app); + ~Worker(); + + /** + * @brief If Kuni is sleeping, this function wake ups her. + */ + void wakeUpIfSleeping() { + mWakeUp = true; + } + + AFuture<> diaryDumpMessages(); + +private: + size_t mName; + ALogger mLogger{"kuni_worker{}.log"_format(mName)}; + AppBase& mApp; + ASet mWorkerPins; + AFuture<> mCoroutine; + std::shared_ptr mAliveToken = std::make_shared(true); + bool mWakeUp = false; + bool mAskCalledThisTurn = false; + aui::float_within_0_1 mRelevanceThreshold = 0.5f; + + IOpenAIChat::Session mTemporaryContext = [this] { + IOpenAIChat::Session s; + s.sessionId = "kuni_main_coro({})"_format(mName); + return s; + }(); + + AFuture<> handleNotification(std::shared_ptr alive, NotificationManager::Notification notification); + + AString takeDiaryEntry(const Diary::EntryExAndRelatedness& i); + + void updateTools(OpenAITools& tools); +}; \ No newline at end of file diff --git a/src/config.cpp b/src/config.cpp index bdb38a9..fbc947e 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -413,6 +413,11 @@ static const std::unordered_map CONFIG_COMMENTS = { "misc.can_leave_chats", "Whether Kuni chat leave a telegram group chat/channel.", }, + { + "misc.worker_count", + "Amount of Kuni's subpersons that process messages. 1 is totally fine.\n" + "This was implemented as a countermeasure to spamming Kuni's DM.", + }, }; static constexpr auto CONFIG_TOML = "config.toml"; @@ -628,6 +633,11 @@ const Config& config() { } *watcher ^ gConfigUpdated(); }); +#ifdef AUI_TESTS_MODULE + // defaults for unit tests + cfg.workerCount = 1; + cfg.antiRepeatMaxHistory = 32; +#endif }; return cfg; } diff --git a/src/config.h b/src/config.h index 7f59f8f..ea2c7a1 100644 --- a/src/config.h +++ b/src/config.h @@ -45,6 +45,7 @@ X(::Config::LockdownMode, chatNotificationFilter, ::Config::LockdownMode::NONE, "misc.chat_notification_filter") \ X(bool, canJoinChats, false, "misc.can_join_chats") \ X(bool, canLeaveChats, true, "misc.can_leave_chats") \ + X(size_t, workerCount, 1, "misc.worker_count") \ X(bool, capabilityWebSearch, false, "capabilities.web_search.enabled") \ X(AString, webSearchOllamaKey, "", "capabilities.web_search.ollama_bearer_key") \ X(bool, capabilityVision, false, "capabilities.vision.enabled") \ diff --git a/src/llmui/telegram.cpp b/src/llmui/telegram.cpp index 640ec6d..a606be5 100644 --- a/src/llmui/telegram.cpp +++ b/src/llmui/telegram.cpp @@ -414,6 +414,10 @@ AFuture llmui::formatChatHistoryMessage( if (!senderName.empty()) { formattedXmlTag += " sender=\"{}\""_format(senderName); } + if (chat.type_->get_id() == td::td_api::chatTypeBasicGroup::ID || chat.type_->get_id() == td::td_api::chatTypeSupergroup::ID) { + // for bans. + formattedXmlTag += " sender_id=\"{}\""_format(senderId); + } } if (msg.interaction_info_) { if (msg.interaction_info_->reactions_) { diff --git a/src/main.cpp b/src/main.cpp index fb4cede..c285a19 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -61,6 +61,9 @@ #include #include #include "tools/ask.h" +#include "tools/group_admin_ban_user.h" +#include "tools/group_admin_remove_message.h" +#include "tools/group_admin_set_user_tag.h" #include "tools/remove_message.h" #include @@ -94,7 +97,6 @@ class App : public AppBase { [[nodiscard]] _ telegram() const { return mTelegram; } -protected: void onOffline() override { mCurrentlyOpenedChat.reset(); setOnline(false); @@ -148,8 +150,8 @@ class App : public AppBase { mCurrentlyOpenedChat->chat->id_, {}, {}, ITelegramClient::toPtr(td::td_api::chatActionTyping())))); } - void updateTools(OpenAITools& actions) override { - AppBase::updateTools(actions); + void updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext) override { + AppBase::updateTools(actions, temporaryContext); if (config().capabilityTakePhoto) { actions.insert(tools::takePhoto(_new(), openAI())); } @@ -158,8 +160,8 @@ class App : public AppBase { } actions.insert(tools::getTelegramChats(telegram(), openAI(), isActingProactively())); actions.insert(tools::searchChats(telegram())); - actions.insert(tools::searchMessages(telegram(), openAI(), temporaryContext())); - actions.insert(tools::viewMessagesAround(telegram(), openAI(), temporaryContext())); + actions.insert(tools::searchMessages(telegram(), openAI(), temporaryContext)); + actions.insert(tools::viewMessagesAround(telegram(), openAI(), temporaryContext)); actions.insert(tools::removeAndBanChat(telegram())); actions.insert({ .name = "open_chat_by_id", @@ -189,7 +191,7 @@ class App : public AppBase { co_return "No such chat"; } - co_return co_await llmuiOpenTelegramChat(ctx.tools, chatId); + co_return co_await llmuiOpenTelegramChat(ctx.logger, ctx.tools, chatId, ctx.temporaryContext); }, }); if (config().canJoinChats) { @@ -219,7 +221,7 @@ class App : public AppBase { co_return "Error: failed to join chat by invite link: {}"_format(e.getMessage()); } - co_return co_await llmuiOpenTelegramChat(ctx.tools, chatId); + co_return co_await llmuiOpenTelegramChat(ctx.logger, ctx.tools, chatId, ctx.temporaryContext); }, }); } @@ -229,8 +231,29 @@ class App : public AppBase { } } - AFuture onCleanContext() override { - AString result = co_await AppBase::onCleanContext(); + AFuture<> sendNotificationsOnInit() { + if (!config().checkChatsOnStartup) { + co_return; + } + // tdlib does not send notifications for unread chats on program startup. we'll fix this. + auto chats = co_await getChats(); + chats |= ranges::actions::reverse; // older first, newest last + for (auto& chat : chats) { + if (chat->unread_count_ == 0) { + continue; + } + // make up a updateNewMessage event and pass it to handleTelegramEvent. the latter will format a + // notification for us. + auto notification = _new(); + notification->message_ = std::move(chat->last_message_); + co_await handleTelegramEvent(std::move(notification)); + } + } + + +protected: + AString onCleanContext() const override { + AString result = AppBase::onCleanContext(); // Alex2772 (9 Jul 2026): // consumes too much content. // if llm wants to share a sticker, it would call get_stickers(). @@ -243,12 +266,7 @@ class App : public AppBase { // result += "\n"; // } // } - co_return result; - } - - AFuture<> onBeforeMainLoop() override { - co_await telegram()->waitForConnection(); - co_await sendNotificationsOnInit(); + return result; } private: @@ -273,25 +291,6 @@ class App : public AppBase { co_return co_await chatIdsToChats(chatList->chat_ids_); } - AFuture<> sendNotificationsOnInit() { - if (!config().checkChatsOnStartup) { - co_return; - } - // tdlib does not send notifications for unread chats on program startup. we'll fix this. - auto chats = co_await getChats(); - chats |= ranges::actions::reverse; // older first, newest last - for (auto& chat : chats) { - if (chat->unread_count_ == 0) { - continue; - } - // make up a updateNewMessage event and pass it to handleTelegramEvent. the latter will format a - // notification for us. - auto notification = _new(); - notification->message_ = std::move(chat->last_message_); - co_await handleTelegramEvent(std::move(notification)); - } - } - template Object> AFuture<> handleTelegramEvent(_ u) { TelegramClientImpl::StubHandler {}(*u); @@ -380,35 +379,36 @@ class App : public AppBase { "\n\n" "You don't have any chat open. Use #open tool to open the chat"; - const bool isImportant = [&] { + const int priority = [&] { if (userId == config().papikChatId) { - return true; + return 1000; } if (config().wakeUpOnPinnedChat) { for (const auto& position : chat->positions_) { if (position->is_pinned_) { - return true; + return 100; } } } - return false; + return 0; }(); - passNotificationToAI( - std::move(notification), - { + notificationManager().passNotificationToAI(NotificationManager::Notification{ + .message = std::move(notification), + .actions = { { .name = "open", .description = "Open \"{}\" chat. Use this if you'd like to reply or see messages."_format(chat->title_), .handler = [this, chatId = chat->id_](OpenAITools::Ctx ctx) -> AFuture { - return llmuiOpenTelegramChat(ctx.tools, chatId); + return llmuiOpenTelegramChat(ctx.logger, ctx.tools, chatId, ctx.temporaryContext); }, }, - }, - isImportant); + .priority = priority, + .pin = ""_format(chat->id_), + }); - if (isImportant) { + if (priority > 0) { wakeUpIfSleeping(); } @@ -434,10 +434,10 @@ class App : public AppBase { AOptional mCurrentlyOpenedChat; public: - AFuture llmuiOpenTelegramChat(OpenAITools& tools, int64_t chatId) { + AFuture llmuiOpenTelegramChat(ALogger& logger, OpenAITools& tools, int64_t chatId, const IOpenAIChat::Session& temporaryContext) { // Check lockdown mode - only allow PAPIK_CHAT_ID if lockdown is enabled if (!co_await util::isAccessibleFromLockdown(*telegram(), chatId)) { - ALogger::err(LOG_TAG) << "Error: Lockdown mode is enabled. You can only open chat with ID {} (PAPIK_CHAT_ID)."_format( + logger.err(LOG_TAG) << "Error: Lockdown mode is enabled. You can only open chat with ID {} (PAPIK_CHAT_ID)."_format( config().papikChatId); co_return "No such chat"; } @@ -445,7 +445,7 @@ class App : public AppBase { co_await telegram()->waitForConnection(); setOnline(); mTelegram->sendQuery(ITelegramClient::toPtr(td::td_api::openChat(chatId))); - removeNotifications("\n"_format(chatId)); + notificationManager().removeNotifications("\n"_format(chatId)); _ chat = co_await mTelegram->getChat(chatId); mCurrentlyOpenedChat.emplace(*this, chat); @@ -473,7 +473,7 @@ class App : public AppBase { const auto msgFormatting = R"(id_); length += to_string(msg->content_).length(); messages.push_back(std::move(msg)); - if (ranges::any_of(temporaryContext(), [&](const IOpenAIChat::Message& msg) { + if (ranges::any_of(temporaryContext, [&](const IOpenAIChat::Message& msg) { return msg.content.contains(msgFormatting); })) { // this message is already in context, which means we don't need to load further. @@ -490,7 +490,7 @@ class App : public AppBase { } } }(); - ALOG_DEBUG(LOG_TAG) << "Loaded " << messages.size() << " message(s): " << chat->title_; + logger.info(LOG_TAG) << "Loaded " << messages.size() << " message(s): " << chat->title_; // Compute response-time metadata for Prometheus. messages[0] is the most recent. mLastOpenedChatLastMessageTime = [&]() -> AOptional { @@ -533,10 +533,11 @@ class App : public AppBase { } // goto naxyi; } + bool isAdmin = false; { for (auto& msg : messages | ranges::view::reverse) { auto msgFormatted = - co_await llmui::formatChatHistoryMessage(*telegram(), *msg, *chat, *openAI(), temporaryContext()); + co_await llmui::formatChatHistoryMessage(*telegram(), *msg, *chat, *openAI(), temporaryContext); for (const auto& i : chatHistoryMessageProcessors) { msgFormatted = co_await i->processChatHistoryMessage(*chat, *msg, std::move(msgFormatted)); } @@ -608,6 +609,10 @@ class App : public AppBase { case td::td_api::chatMemberStatusBanned::ID: isMember = false; break; + case td::td_api::chatMemberStatusCreator::ID: + case td::td_api::chatMemberStatusAdministrator::ID: + isAdmin = true; + break; default: break; } @@ -688,9 +693,8 @@ Do NOT forward ads, sponsored posts, or low-value content. naxyi: tools = OpenAITools { - tools::sendTelegramMessage( - telegram(), openAI(), chat, _new>>(std::move(messages))), - tools::getChatPhoto(telegram(), openAI(), chat, temporaryContext()), + tools::sendTelegramMessage(telegram(), openAI(), chat, _new>>(std::move(messages))), + tools::getChatPhoto(telegram(), openAI(), chat, temporaryContext), tools::reactWithEmoji(telegram(), chat), tools::removeMessage(telegram(), chat), tools::editMessageText(telegram(), chat), @@ -701,15 +705,21 @@ Do NOT forward ads, sponsored posts, or low-value content. tools.insert(tools::stickers::send(telegram(), chat)); } - if (config().canLeaveChats) { - switch (chat->type_->get_id()) { - case td::td_api::chatTypeBasicGroup::ID: - case td::td_api::chatTypeSupergroup::ID: + switch (chat->type_->get_id()) { + case td::td_api::chatTypeBasicGroup::ID: + case td::td_api::chatTypeSupergroup::ID: + + if (config().canLeaveChats) { tools.insert(tools::leaveChat(telegram(), chat)); - break; - default: - break; - } + } + if (isAdmin) { + tools.insert(tools::groupAdminRemoveMessage(telegram(), chat)); + tools.insert(tools::groupAdminBanUser(telegram(), chat)); + tools.insert(tools::groupAdminSetUserTag(telegram(), chat)); + } + break; + default: + break; } co_return result; @@ -750,6 +760,7 @@ AUI_ENTRY { AObject::connect(telegram->loggedIn, telegram, [&] { auto openAI = _new(std::make_unique()); app = _new(telegram, openAI); + async << app->sendNotificationsOnInit(); if (config().proxyEnabled) { auto diary = std::make_shared(Diary::Init { .diaryDir = "data/diary", .openAI = openAI }); @@ -804,7 +815,9 @@ AUI_ENTRY { IEventLoop::Handle h(&gEventLoop); gEventLoop.loop(); - if (app) { async << app->diaryDumpMessages(); } + if (app) { + async << app->diaryDumpMessages(); + } if (contextBridge) { async << contextBridge->collectAndSaveSessionsNotNewerThan(std::chrono::system_clock::now()); } while (!async.empty()) { gEventLoop.iteration(); } diff --git a/src/tools/ask.cpp b/src/tools/ask.cpp index f1806d5..3c292a0 100644 --- a/src/tools/ask.cpp +++ b/src/tools/ask.cpp @@ -120,7 +120,6 @@ static AFuture ask(IOpenAIChat& openAI, Diary& diary, const AString& qu .content = "\n{}\n\n\n{}"_format(prompts().characterBase, query), }, }; - messages.sessionId = "ask"; bool toolCallHappened = false; @@ -168,7 +167,7 @@ Do not make up facts. Rely exclusively on provided context. } OpenAITools::Tool -tools::ask(std::function additionalDetails, _ openAI, Diary& diary) { +tools::ask(std::function additionalDetails, _ openAI, Diary diary) { return { .name = "ask", .description = "Consult with Kuni's main knowledge database and the internet (subagent). Use this to " @@ -193,7 +192,7 @@ tools::ask(std::function additionalDetails, _ openAI, Di }, .required = {"query"}, }, - .handler = [additionalDetails, openAI, &diary](OpenAITools::Ctx ctx) -> AFuture { + .handler = [additionalDetails, openAI, diary = std::move(diary)](OpenAITools::Ctx ctx) -> AFuture { auto query = ctx.args["query"].asStringOpt().valueOrException("\"query\" string is required"); if (query.length() < 10) { // Alex2772 16-04-2026: @@ -224,7 +223,7 @@ tools::ask(std::function additionalDetails, _ openAI, Di "- how can I improve my reaction?\n" "- {}"_format(details, query); } - co_return (co_await ask(*openAI, diary, query, {.confidenceFactor = 0.f})); + co_return (co_await ask(*openAI, const_cast(diary), query, {.confidenceFactor = 0.f})); }, }; } diff --git a/src/tools/ask.h b/src/tools/ask.h index 60ba02a..e5103f0 100644 --- a/src/tools/ask.h +++ b/src/tools/ask.h @@ -6,5 +6,5 @@ #include namespace tools { -OpenAITools::Tool ask(std::function additionalDetails, _ openAI, Diary& diary); +OpenAITools::Tool ask(std::function additionalDetails, _ openAI, Diary diary); } diff --git a/src/tools/group_admin_ban_user.cpp b/src/tools/group_admin_ban_user.cpp new file mode 100644 index 0000000..95b1904 --- /dev/null +++ b/src/tools/group_admin_ban_user.cpp @@ -0,0 +1,44 @@ +// +// Created by alex2772 on 6/18/26. +// + +#include "group_admin_ban_user.h" + +#include "util/json_utils.h" + +OpenAITools::Tool tools::groupAdminBanUser(_ telegram, _ chat) { + return { + .name = "group_admin_ban_user", + .description = "Administrative tool for \"{}\" chat. Permanently deletes the specified user from the group " + "chat. Use this if you consistently feel this user is offensive to you, other people or " + "otherwise not worth the time (e.g., is consistently rude, toxic, or the " + "user brings no value to others). This is a serious, irreversible decision - use #ask " + "beforehand to double check your reasoning if unsure. This tool is only available in " + "group chats; for a private chat/DM with a specific person, use #remove_and_ban_chat " + "instead"_format(chat->title_), + .parameters = { + .properties = { + {"user_id", {.type = "integer", .description = "ID of the user to remove. Can be acquired from sender_id."}}, + }, + .required = {"user_id" }, + }, + .handler = [telegram = std::move(telegram), chat = std::move(chat)](OpenAITools::Ctx ctx) -> AFuture { + if (ctx.args.contains("chat_id")) { + if (ctx.args["chat_id"].asLongInt() != chat->id_) { + co_return "Error: you can't ban users from other chats. Open them first. You are currently in chat \"{}\""_format(chat->title_); + } + } + if (!ctx.args.contains("user_id")) { + throw AException("user_id is a mandatory argument"); + } + const auto targetUserId = util::jsonAsLongInt(ctx.args["user_id"]).valueOrException("user_id"); + + auto user = co_await telegram->getUser(targetUserId); + const auto result = "User {} {} were banned successfully from \"{}\" chat."_format(user->first_name_, user->last_name_, chat->title_); + + auto ok = co_await telegram->sendQueryWithResult(ITelegramClient::toPtr(td::td_api::banChatMember( + chat->id_, ITelegramClient::toPtr(td::td_api::messageSenderUser(user->id_)), 0, false))); + co_return result; + }, + }; +} diff --git a/src/tools/group_admin_ban_user.h b/src/tools/group_admin_ban_user.h new file mode 100644 index 0000000..b40963f --- /dev/null +++ b/src/tools/group_admin_ban_user.h @@ -0,0 +1,7 @@ +#pragma once +#include "OpenAITools.h" +#include "telegram/ITelegramClient.h" + +namespace tools { +OpenAITools::Tool groupAdminBanUser(_ telegram, _ chat); +} diff --git a/src/tools/group_admin_remove_message.cpp b/src/tools/group_admin_remove_message.cpp new file mode 100644 index 0000000..70c2465 --- /dev/null +++ b/src/tools/group_admin_remove_message.cpp @@ -0,0 +1,50 @@ +// +// Created by alex2772 on 6/18/26. +// + +#include "group_admin_remove_message.h" + +#include "util/json_utils.h" + +OpenAITools::Tool tools::groupAdminRemoveMessage(_ telegram, _ chat) { + return { + .name = "group_admin_remove_message", + .description = "Administrative tool for \"{}\" chat. Deletes specified message for both you and participant(s).\n" + "You should use this to remove spam/inappropriate message(s) in group chats."_format(chat->title_), + .parameters = { + .properties = { + {"message_id", {.type = "integer|array", .description = "ID(s) of the message(s) to delete. Taken from message_id attribute in tag."}}, + }, + .required = {"message_id" }, + }, + .handler = [telegram = std::move(telegram), chat = std::move(chat)](OpenAITools::Ctx ctx) -> AFuture { + if (ctx.args.contains("chat_id")) { + if (ctx.args["chat_id"].asLongInt() != chat->id_) { + co_return "Error: you can't remove messages from other chats. Open them first. You are currently in chat \"{}\""_format(chat->title_); + } + } + if (!ctx.args.contains("message_id")) { + throw AException("message_id is a mandatory argument"); + } + + td::td_api::array messages; + if (ctx.args["message_id"].isArray()) { + for (const auto& i : ctx.args["message_id"].asArray()) { + messages.push_back(util::jsonAsLongInt(i).valueOrException("expected integer")); + } + } else { + messages.push_back(util::jsonAsLongInt(ctx.args["message_id"]).valueOrException("message_id")); + } + + const auto result = "Messages {} were deleted successfully."_format(std::span(messages)); + + for (auto& i : messages) { + // remap client-side messageId (which was reported to llm to server-side messageId) + i = (co_await telegram->getMessage(chat->id_, i))->id_; + } + + auto ok = co_await telegram->sendQueryWithResult(ITelegramClient::toPtr(td::td_api::deleteMessages(chat->id_, std::move(messages), true))); + co_return result; + }, + }; +} diff --git a/src/tools/group_admin_remove_message.h b/src/tools/group_admin_remove_message.h new file mode 100644 index 0000000..118deac --- /dev/null +++ b/src/tools/group_admin_remove_message.h @@ -0,0 +1,7 @@ +#pragma once +#include "OpenAITools.h" +#include "telegram/ITelegramClient.h" + +namespace tools { +OpenAITools::Tool groupAdminRemoveMessage(_ telegram, _ chat); +} diff --git a/src/tools/group_admin_set_user_tag.cpp b/src/tools/group_admin_set_user_tag.cpp new file mode 100644 index 0000000..de58ebb --- /dev/null +++ b/src/tools/group_admin_set_user_tag.cpp @@ -0,0 +1,52 @@ +// +// Created by alex2772 on 6/18/26. +// + +#include "group_admin_set_user_tag.h" + +#include "util/json_utils.h" + +OpenAITools::Tool tools::groupAdminSetUserTag(_ telegram, _ chat) { + return { + .name = "group_admin_set_user_tag", + .description = "Administrative tool for \"{}\" chat. Sets (or clears) the member tag shown next to a " + "user's name on every message they send in this group. Telegram member tags are normally " + "used to pin a short label to someone — their role, what they're into, a running joke, or " + "just a nickname that sticks. Unlike a one-off insult in chat, a tag is persistent and " + "visible on literally everything that person says afterwards, so this is a real vibe-check: " + "only tag someone if you've actually formed an opinion about them worth broadcasting " + "permanently, not on a whim. Set a tag based on your perception of the user (use #ask). Max 16" + "characters, no emoji. Pass an empty tag to remove an existing one."_format(chat->title_), + .parameters = { + .properties = { + {"user_id", {.type = "integer", .description = "ID of the user to tag. Can be acquired from sender_id."}}, + {"tag", {.type = "string", .description = "New tag text, up to 16 characters, no emoji. Pass an empty string to remove the current tag."}}, + }, + .required = {"user_id", "tag"}, + }, + .handler = [telegram = std::move(telegram), chat = std::move(chat)](OpenAITools::Ctx ctx) -> AFuture { + if (ctx.args.contains("chat_id")) { + if (ctx.args["chat_id"].asLongInt() != chat->id_) { + co_return "Error: you can't tag users from other chats. Open them first. You are currently in chat \"{}\""_format(chat->title_); + } + } + if (!ctx.args.contains("user_id")) { + throw AException("user_id is a mandatory argument"); + } + if (!ctx.args.contains("tag")) { + throw AException("tag is a mandatory argument"); + } + const auto targetUserId = util::jsonAsLongInt(ctx.args["user_id"]).valueOrException("user_id"); + const auto tag = ctx.args["tag"].asStringOpt().valueOrException("tag string is required"); + + auto user = co_await telegram->getUser(targetUserId); + const auto result = tag.empty() + ? "Tag for user {} {} was removed successfully in \"{}\" chat."_format(user->first_name_, user->last_name_, chat->title_) + : "User {} {} was tagged as \"{}\" successfully in \"{}\" chat."_format(user->first_name_, user->last_name_, tag, chat->title_); + + auto ok = co_await telegram->sendQueryWithResult( + ITelegramClient::toPtr(td::td_api::setChatMemberTag(chat->id_, user->id_, tag.toStdString()))); + co_return result; + }, + }; +} diff --git a/src/tools/group_admin_set_user_tag.h b/src/tools/group_admin_set_user_tag.h new file mode 100644 index 0000000..9f4c6bd --- /dev/null +++ b/src/tools/group_admin_set_user_tag.h @@ -0,0 +1,11 @@ +// +// Created by alex2772 on 6/18/26. +// + +#pragma once +#include "OpenAITools.h" +#include "telegram/ITelegramClient.h" + +namespace tools { +OpenAITools::Tool groupAdminSetUserTag(_ telegram, _ chat); +} diff --git a/src/tools/remove_message.cpp b/src/tools/remove_message.cpp index 2c7794e..997b4b8 100644 --- a/src/tools/remove_message.cpp +++ b/src/tools/remove_message.cpp @@ -10,7 +10,7 @@ OpenAITools::Tool tools::removeMessage(_ telegram, _title_), + "You should use this when you mistakenly sent a message to a wrong chat."_format(chat->title_), .parameters = { .properties = { {"message_id", {.type = "integer|array", .description = "ID(s) of the message(s) to delete. Taken from message_id attribute in tag."}}, @@ -20,7 +20,7 @@ OpenAITools::Tool tools::removeMessage(_ telegram, _ AFuture { if (ctx.args.contains("chat_id")) { if (ctx.args["chat_id"].asLongInt() != chat->id_) { - co_return "Error: you can't send messages to other chats. Open them first. You are currently in chat \"{}\""_format(chat->title_); + co_return "Error: you can't remove messages from other chats. Open them first. You are currently in chat \"{}\""_format(chat->title_); } } if (!ctx.args.contains("message_id")) { diff --git a/tests/AppBaseUnitTest.cpp b/tests/AppBaseUnitTest.cpp deleted file mode 100644 index 8456523..0000000 --- a/tests/AppBaseUnitTest.cpp +++ /dev/null @@ -1,499 +0,0 @@ -// -// Created by alex2772 on 5/13/26. -// - -#include "AppBase.h" -#include "IOpenAIChat.h" -#include "OpenAIMock.h" -#include "OpenAITools.h" -#include "Diary.h" -#include "common.h" - -#include -#include -#include -#include -#include - -#include - -// ============================================================================ -// Helper: create a minimal chat response that calls #wait (pause) -// ============================================================================ -static _ makeWaitResponse() { - IOpenAIChat::Message msg; - msg.role = IOpenAIChat::Message::Role::ASSISTANT; - msg.content = ""; - msg.tool_calls = { - IOpenAIChat::Message::ToolCall{ - .id = "call_wait_1", - .index = 0, - .type = "function", - .function = { - .name = "wait", - .arguments = "{}", - }, - }, - }; - - auto result = _new(); - result->response.raw = { - .choices = { - IOpenAIChat::Response::Choice{ - .index = 0, - .message = std::move(msg), - .finish_reason = "tool_calls", - }, - }, - .usage = { .prompt_tokens = 10, .completion_tokens = 5, .total_tokens = 15 }, - }; - result->completed.supplyValue(); - return result; -} - -// ============================================================================ -// Helper: create an embedding result (dummy vector) -// ============================================================================ -static std::valarray makeDummyEmbedding() { - return std::valarray{0.1, 0.2, 0.3, 0.4, 0.5}; -} - -// ============================================================================ -// AppTestHarness — controlled AppBase subclass for unit testing -// ============================================================================ -class AppTestHarness : public AppBase { -public: - explicit AppTestHarness(_ openAI) - : AppBase(Init{ - .workingDir = "test_data_appbase_unit", - .openAI = std::move(openAI), - }) - { - // Clean slate - APath("test_data_appbase_unit").removeFileRecursive(); - } - - ~AppTestHarness() override { - APath("test_data_appbase_unit").removeFileRecursive(); - } - - // Expose protected members for testing - using AppBase::mTemporaryContext; - using AppBase::mRelevanceThreshold; - using AppBase::openAI; - using AppBase::takeDiaryEntry; - using AppBase::removeNotifications; - using AppBase::updateTools; - using AppBase::diaryDumpMessages; - using AppBase::onBeforeMainLoop; - - // Expose diary - using AppBase::diary; - - // Count how many times updateTools was called - int updateToolsCallCount = 0; - - void updateTools(OpenAITools& actions) override { - ++updateToolsCallCount; - AppBase::updateTools(actions); - - // Always provide #wait and #pause so the main loop can terminate - actions.insert({ - .name = "pause", - .description = "Pauses the conversation", - .handler = [](OpenAITools::Ctx) -> AFuture { - co_return "Paused"; - }, - }); - actions.insert({ - .name = "wait", - .description = "Wait until further notifications", - .handler = [](OpenAITools::Ctx) -> AFuture { - co_return "Waiting"; - }, - }); - } -}; - -// ============================================================================ -// Test fixture -// ============================================================================ -class AppBaseUnitTest : public ::testing::Test { -protected: - void SetUp() override { - APath("test_data_appbase_unit").removeFileRecursive(); - } - - void TearDown() override { - APath("test_data_appbase_unit").removeFileRecursive(); - } - -}; - -// ============================================================================ -// passNotificationToAI — basic queue and signal -// ============================================================================ -TEST_F(AppBaseUnitTest, PassNotificationToAIBasic) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _new(); - EXPECT_CALL(*openAI.get(), chatStreaming(::testing::_, ::testing::_)).WillOnce(::testing::Return(makeWaitResponse())); - - AppTestHarness app(openAI); - async << app.passNotificationToAI("Test notification message").onProcessed; - while (!async.empty()) { - loop.iteration(); - } - - // The notification should have been processed — context is non-empty - EXPECT_FALSE(app.temporaryContext().empty()); -} - -// ============================================================================ -// passNotificationToAI — multiple notifications are queued -// ============================================================================ -TEST_F(AppBaseUnitTest, PassNotificationToAIMultiple) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _new(); - EXPECT_CALL(*static_cast(openAI.get()), chatStreaming(::testing::_, ::testing::_)) - .WillOnce(::testing::Return(makeWaitResponse())) - .WillOnce(::testing::Return(makeWaitResponse())) - ; - - AppTestHarness app(openAI); - - async << app.passNotificationToAI("First notification").onProcessed; - while (!async.empty()) { - loop.iteration(); - } - - // After first notification is processed, send another - async << app.passNotificationToAI("Second notification").onProcessed; - while (!async.empty()) { - loop.iteration(); - } - - EXPECT_FALSE(app.temporaryContext().empty()); -} - -// ============================================================================ -// passNotificationToAI — first=true inserts at front -// ============================================================================ -TEST_F(AppBaseUnitTest, PassNotificationToAIFirst) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _new(); - EXPECT_CALL(*openAI.get(), chatStreaming(::testing::_, ::testing::_)).WillOnce(::testing::Return(makeWaitResponse())) - ; - - AppTestHarness app(openAI); - - // Insert urgent first, then normal — urgent should be processed first - async << app.passNotificationToAI("Urgent notification", {}, true).onProcessed; - while (!async.empty()) { - loop.iteration(); - } - - EXPECT_FALSE(app.temporaryContext().empty()); -} - -// ============================================================================ -// removeNotifications — removes by substring -// ============================================================================ -TEST_F(AppBaseUnitTest, RemoveNotificationsBySubstring) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _new(); - EXPECT_CALL(*static_cast(openAI.get()), chatStreaming(::testing::_, ::testing::_)).WillOnce(::testing::Return(makeWaitResponse())); - - AppTestHarness app(openAI); - - app.passNotificationToAI("Message about cats"); - async << app.passNotificationToAI("Message about dogs").onProcessed; - app.passNotificationToAI("Message about cats again"); - - app.removeNotifications("cats"); - - while (!async.empty()) { - loop.iteration(); - } - - // No crash = success - EXPECT_TRUE(true); -} - -// ============================================================================ -// removeNotifications — no match does nothing -// ============================================================================ -TEST_F(AppBaseUnitTest, RemoveNotificationsNoMatch) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _new(); - EXPECT_CALL(*static_cast(openAI.get()), chatStreaming(::testing::_, ::testing::_)) - .WillOnce(::testing::Return(makeWaitResponse())) - .WillOnce(::testing::Return(makeWaitResponse())) - ; - - AppTestHarness app(openAI); - - async << app.passNotificationToAI("Message about cats").onProcessed; - async << app.passNotificationToAI("Message about dogs").onProcessed; - - app.removeNotifications("nonexistent"); - - while (!async.empty()) { - loop.iteration(); - } - - EXPECT_TRUE(true); // no crash = success -} - -// ============================================================================ -// takeDiaryEntry — formats entry with XML tags -// ============================================================================ -TEST_F(AppBaseUnitTest, TakeDiaryEntryFormatsCorrectly) { - auto openAI = _cast(_new()); - AppTestHarness app(openAI); - - // Manually add a diary entry - APath("test_data_appbase_unit/diary").makeDirs(); - app.diary().save(Diary::EntryEx{ - .id = "test_entry_1", - .metadata = { - .score = 0.5f, - .embedding = makeDummyEmbedding(), - }, - .freeformBody = "John likes pizza and programming.", - }); - app.diary().reload(); - - // Query to get an EntryExAndRelatedness - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - Diary::EntryExAndRelatedness found{}; - bool foundEntry = false; - - async << [&]() -> AFuture<> { - auto results = co_await app.diary().query(makeDummyEmbedding(), {}); - if (!results.empty()) { - found = results.front(); - foundEntry = true; - } - }(); - - while (!async.empty()) { - loop.iteration(); - } - - ASSERT_TRUE(foundEntry); - - // takeDiaryEntry should format with XML tags - AString formatted = app.takeDiaryEntry(found); - EXPECT_FALSE(formatted.empty()); - EXPECT_TRUE(formatted.contains("(_new()); - AppTestHarness app(openAI); - - // Put the entry text into temporary context first - app.mTemporaryContext << IOpenAIChat::Message{ - .role = IOpenAIChat::Message::Role::USER, - .content = "John likes pizza and programming.", - }; - - // Save the same text as a diary entry - APath("test_data_appbase_unit/diary").makeDirs(); - app.diary().save(Diary::EntryEx{ - .id = "test_entry_dup", - .metadata = { - .score = 0.5f, - .embedding = makeDummyEmbedding(), - }, - .freeformBody = "John likes pizza and programming.", - }); - app.diary().reload(); - - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - Diary::EntryExAndRelatedness found{}; - bool foundEntry = false; - - async << [&]() -> AFuture<> { - auto results = co_await app.diary().query(makeDummyEmbedding(), {}); - if (!results.empty()) { - found = results.front(); - foundEntry = true; - } - }(); - - while (!async.empty()) { - loop.iteration(); - } - - ASSERT_TRUE(foundEntry); - - // takeDiaryEntry should return empty because the content is already in context - AString formatted = app.takeDiaryEntry(found); - EXPECT_TRUE(formatted.empty()); -} - -// ============================================================================ -// takeDiaryEntry — increments usage count and updates score -// ============================================================================ -TEST_F(AppBaseUnitTest, TakeDiaryEntryUpdatesMetadata) { - auto openAI = _cast(_new()); - AppTestHarness app(openAI); - - APath("test_data_appbase_unit/diary").makeDirs(); - app.diary().save(Diary::EntryEx{ - .metadata = { - .score = 0.0f, - .usageCount = 0, - .embedding = makeDummyEmbedding(), - }, - .freeformBody = "Unique content about space exploration.", - }); - app.diary().reload(); - - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - Diary::EntryExAndRelatedness found{}; - bool foundEntry = false; - - async << [&]() -> AFuture<> { - auto results = co_await app.diary().query(makeDummyEmbedding(), {}); - if (!results.empty()) { - found = results.front(); - foundEntry = true; - } - }(); - - while (!async.empty()) { - loop.iteration(); - } - - ASSERT_TRUE(foundEntry); - - [[maybe_unused]] auto entry = app.takeDiaryEntry(found); - - // After takeDiaryEntry, the entry is unloaded (removed from cache), - // so we can't check the in-memory metadata. But we can verify - // the entry was removed from the diary listing. - EXPECT_FALSE(ranges::any_of(app.diary().list(), [](const auto& e) { - return e.id == "test_entry_meta"; - })); -} - -// ============================================================================ -// updateTools — adds unified ask tool -// ============================================================================ -TEST_F(AppBaseUnitTest, UpdateToolsAddsExpectedTools) { - auto openAI = _cast(_new()); - AppTestHarness app(openAI); - - OpenAITools tools{}; - app.updateTools(tools); - - auto handlers = tools.handlers(); - EXPECT_TRUE(handlers.contains("ask")); - AThread::processMessages(); -} - -// ============================================================================ -// updateTools — called during notification processing -// ============================================================================ -TEST_F(AppBaseUnitTest, UpdateToolsCalledDuringProcessing) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _cast(_new()); - EXPECT_CALL(*static_cast(openAI.get()), chatStreaming(::testing::_, ::testing::_)) - .WillOnce(::testing::Return(makeWaitResponse())); - - AppTestHarness app(openAI); - - int beforeCount = app.updateToolsCallCount; - async << app.passNotificationToAI("Test").onProcessed; - while (!async.empty()) { - loop.iteration(); - } - - // updateTools should have been called at least once more - EXPECT_GE(app.updateToolsCallCount, beforeCount); -} - -// ============================================================================ -// isActingProactively — initially false -// ============================================================================ -TEST_F(AppBaseUnitTest, IsActingProactivelyInitiallyFalse) { - auto openAI = _cast(_new()); - AppTestHarness app(openAI); - - EXPECT_FALSE(app.isActingProactively()); -} - -// ============================================================================ -// temporaryContext — initially empty -// ============================================================================ -TEST_F(AppBaseUnitTest, TemporaryContextInitiallyEmpty) { - auto openAI = _cast(_new()); - AppTestHarness app(openAI); - - EXPECT_TRUE(app.temporaryContext().empty()); - AThread::processMessages(); -} - -// ============================================================================ -// temporaryContext — accumulates messages after notification -// ============================================================================ -TEST_F(AppBaseUnitTest, TemporaryContextAccumulatesMessages) { - AAsyncHolder async; - AEventLoop loop; - IEventLoop::Handle h(&loop); - - auto openAI = _cast(_new()); - EXPECT_CALL(*static_cast(openAI.get()), chatStreaming(::testing::_, ::testing::_)) - .WillOnce(::testing::Return(makeWaitResponse())); - - AppTestHarness app(openAI); - - async << app.passNotificationToAI("Hello from test").onProcessed; - while (!async.empty()) { - loop.iteration(); - } - - // After processing, the context should have at least the user message - // and the assistant response - EXPECT_FALSE(app.temporaryContext().empty()); - - // The last message should be from the assistant (tool call response) - const auto& lastMsg = app.temporaryContext().last(); - EXPECT_EQ(lastMsg.role, IOpenAIChat::Message::Role::TOOL); -} diff --git a/tests/ContextBridgeUnitTest.cpp b/tests/ContextBridgeUnitTest.cpp index 9b0a439..c62c591 100644 --- a/tests/ContextBridgeUnitTest.cpp +++ b/tests/ContextBridgeUnitTest.cpp @@ -136,14 +136,6 @@ class MockDiary : public Diary { // called by ContextBridge directly (summarization goes via makeHttpRequest), // so we only need the diary's save/reload API. explicit MockDiary(const APath& dir) : Diary(Init{.diaryDir = dir, .openAI = _new()}) {} - - // Track save calls for assertions. - AVector savedEntries; - - void save(const EntryEx& entry) override { - savedEntries << entry.freeformBody; - Diary::save(entry); - } }; // ─── Fixture ────────────────────────────────────────────────────────────────── @@ -217,8 +209,6 @@ TEST_F(ContextBridgeTest, SingleSessionFlushed) { ASSERT_EQ(llm.receivedRequests.size(), 1u); // The request must have stream=false (non-streaming mode for summarization) EXPECT_FALSE(llm.receivedRequests.at(0)["stream"].asBoolOpt().valueOr(true)); - // Diary entries must be saved - EXPECT_GE(diary->savedEntries.size(), 1u); } // 3. Updating an existing session (same salt prefix) does NOT create a duplicate. diff --git a/tests/DairyIntegrationTest.cpp b/tests/DairyIntegrationTest.cpp index da165f3..e0c553d 100644 --- a/tests/DairyIntegrationTest.cpp +++ b/tests/DairyIntegrationTest.cpp @@ -21,7 +21,7 @@ TEST(DiaryIntegration, Basic) { AAsyncHolder async; auto app = _new(); - async << app->passNotificationToAI(R"( + async << app->notificationManager().passNotificationToAI({R"( Today you read an article. Contents below. The source character set of C source programs is contained within the 7-bit ASCII character set but is a superset of the @@ -37,7 +37,7 @@ encourages implementations not to do so. Through C++14, trigraphs are supported Visual C++ continues to support trigraph substitution, but it's disabled by default. For information on how to enable trigraph substitution, see /Zc:trigraphs (Trigraphs Substitution). -)").onProcessed; +)"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -67,12 +67,12 @@ TEST(DiaryIntegration, Remember) { .WillByDefault([](AString text) -> AFuture<> { co_return; }); EXPECT_CALL(*app, telegramPostMessage(testing::_)).Times(testing::AtLeast(1)); - async << app->passNotificationToAI(R"( + async << app->notificationManager().passNotificationToAI({R"( You received a message from Alex2772 (chat_id=1): Today I was playing several games of Dota 2. Both times I was playing Arc Warden and both times we lost :( my teammates weren't bad though. -)").onProcessed; +)"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -93,13 +93,13 @@ Today I was playing several games of Dota 2. Both times I was playing Arc Warden auto app = _new(); testing::InSequence s; bool called = false; - async << app->passNotificationToAI(R"( + async << app->notificationManager().passNotificationToAI({R"( You received a message from Alex2772 (chat_id=1): Today I won a match in Dota 2 Guess which hero I was playing :) -)").onProcessed; +)"}).onProcessed; ON_CALL(*app, telegramPostMessage(testing::_)) .WillByDefault([&](AString text) noexcept -> AFuture<> { const auto lower = text.lowercase(); @@ -263,7 +263,7 @@ TEST(DiaryIntegration, RealWorldChatHistorySneakyTopicSwitch) { EXPECT_CALL(*app, openChat()).Times(testing::AtLeast(1)); EXPECT_CALL(*app, telegramPostMessage(testing::_)).Times(testing::AtLeast(1)); - async << app->passNotificationToAI("You recevied a notification. Please use #open_chat to see mesages.").onProcessed; + async << app->notificationManager().passNotificationToAI({"You recevied a notification. Please use #open_chat to see mesages."}).onProcessed; while (!async.empty()) { loop.iteration(); @@ -341,7 +341,7 @@ TEST(DiaryIntegration, ConversationNoFollowUp) { EXPECT_CALL(*app, openChat()).Times(testing::AtLeast(1)); EXPECT_CALL(*app, telegramPostMessage(testing::_)).Times(testing::Exactly(0)); - async << app->passNotificationToAI("You recevied a notification. Please use #open_chat to see mesages.").onProcessed; + async << app->notificationManager().passNotificationToAI({"You recevied a notification. Please use #open_chat to see mesages."}).onProcessed; while (!async.empty()) { loop.iteration(); diff --git a/tests/NotificationManagerUnitTest.cpp b/tests/NotificationManagerUnitTest.cpp new file mode 100644 index 0000000..2ab7f73 --- /dev/null +++ b/tests/NotificationManagerUnitTest.cpp @@ -0,0 +1,456 @@ +// +// Unit tests for NotificationManager. +// +// Tests: +// 1. Priority ordering — higher-priority notifications are dequeued first +// via nextNotification(). +// 2. Notification removal — removeNotifications() removes by substring. +// 3. Worker pin accumulation — after taking a pinned notification, the +// worker's pin set is updated. +// 4. Unpinned notification preferred over pinned — nextNotification picks +// an unpinned notification before a pinned one owned by another worker. +// 5. Multiple notifications processed in priority order across multiple +// passes of the worker loop. +// + +#include "NotificationManager.h" +#include +#include +#include + +using namespace std::chrono_literals; + +// ============================================================================ +// Fixture +// ============================================================================ +class NotificationManagerUnitTest : public ::testing::Test { +protected: + AEventLoop mLoop; + AAsyncHolder mAsync; + IEventLoop::Handle mLoopHandle{&mLoop}; +}; + +// ============================================================================ +// Helper: construct an empty OpenAITools value. +// ============================================================================ +static OpenAITools emptyActions() { + return OpenAITools({}); +} + +// ============================================================================ +// 1. Priority ordering +// +// Three notifications with priorities 1, 5, 10 are inserted in reverse +// priority order. The worker must receive them in descending priority +// order (10 → 5 → 1), confirming that passNotificationToAI inserts at the +// correct position and nextNotification pops from the front. +// ============================================================================ +TEST_F(NotificationManagerUnitTest, PriorityOrdering) { + NotificationManager manager; + ASet workerPins; + AVector receivedMessages; + + // Worker: collect up to 3 messages then stop. + mAsync << manager.run(workerPins, [&](NotificationManager::Notification notification) -> AFuture { + receivedMessages << notification.message; + co_return receivedMessages.size() < 3; + }); + + // Insert in reverse priority order. + manager.passNotificationToAI({ + .message = "low", + .actions = emptyActions(), + .priority = 1, + }); + manager.passNotificationToAI({ + .message = "high", + .actions = emptyActions(), + .priority = 10, + }); + manager.passNotificationToAI({ + .message = "medium", + .actions = emptyActions(), + .priority = 5, + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } + + ASSERT_EQ(receivedMessages.size(), 3u); + // The message has a timestamp appended, so just check the prefix. + EXPECT_THAT(receivedMessages[0], ::testing::StartsWith("high")); + EXPECT_THAT(receivedMessages[1], ::testing::StartsWith("medium")); + EXPECT_THAT(receivedMessages[2], ::testing::StartsWith("low")); +} + +// ============================================================================ +// 2. Notification removal by substring +// +// removeNotifications removes all notifications whose message contains +// the given substring, without affecting the rest. +// ============================================================================ +TEST_F(NotificationManagerUnitTest, RemoveNotifications) { + NotificationManager manager; + + manager.passNotificationToAI({ + .message = "hello world", + .actions = emptyActions(), + }); + manager.passNotificationToAI({ + .message = "goodbye world", + .actions = emptyActions(), + }); + manager.passNotificationToAI({ + .message = "hello everyone", + .actions = emptyActions(), + }); + + // Remove all that contain "goodbye" + manager.removeNotifications("goodbye"); + + // Verify by having a worker drain the queue + ASet workerPins; + AVector remainingMessages; + + mAsync << manager.run(workerPins, [&](NotificationManager::Notification notification) -> AFuture { + remainingMessages << notification.message; + co_return remainingMessages.size() < 2; + }); + + // Wake up the worker with an unpinned notification + manager.passNotificationToAI({ + .message = "wakeup", + .actions = emptyActions(), + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } + + ASSERT_EQ(remainingMessages.size(), 2u); + EXPECT_THAT(remainingMessages[0], ::testing::StartsWith("hello world")); + EXPECT_THAT(remainingMessages[1], ::testing::StartsWith("hello everyone")); +} + +// ============================================================================ +// 3. Next notification queue order with mixed priorities +// +// After inserting many notifications and letting the worker process them, +// verify that the highest-priority items come out first regardless of +// insertion order. +// ============================================================================ +TEST_F(NotificationManagerUnitTest, QueueOrder) { + NotificationManager manager; + ASet workerPins; + AVector receivedPriorities; + + // Worker: collect up to 5 notifications then stop. + mAsync << manager.run(workerPins, [&](NotificationManager::Notification notification) -> AFuture { + receivedPriorities << notification.priority; + co_return receivedPriorities.size() < 5; + }); + + // Insert in arbitrary order. + manager.passNotificationToAI({ + .message = "a", + .actions = emptyActions(), + .priority = 3, + }); + manager.passNotificationToAI({ + .message = "b", + .actions = emptyActions(), + .priority = 7, + }); + manager.passNotificationToAI({ + .message = "c", + .actions = emptyActions(), + .priority = 1, + }); + manager.passNotificationToAI({ + .message = "d", + .actions = emptyActions(), + .priority = 9, + }); + manager.passNotificationToAI({ + .message = "e", + .actions = emptyActions(), + .priority = 5, + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } + + ASSERT_EQ(receivedPriorities.size(), 5u); + EXPECT_EQ(receivedPriorities[0], 9); + EXPECT_EQ(receivedPriorities[1], 7); + EXPECT_EQ(receivedPriorities[2], 5); + EXPECT_EQ(receivedPriorities[3], 3); + EXPECT_EQ(receivedPriorities[4], 1); +} + +// ============================================================================ +// 4. Equal priorities — FIFO among equal-priority items +// +// Notifications with the same priority should be processed in the order +// they were inserted (FIFO). +// ============================================================================ +TEST_F(NotificationManagerUnitTest, EqualPriorityFifo) { + NotificationManager manager; + ASet workerPins; + AVector receivedMessages; + + mAsync << manager.run(workerPins, [&](NotificationManager::Notification notification) -> AFuture { + receivedMessages << notification.message; + co_return receivedMessages.size() < 3; + }); + + // All same priority — should be FIFO. + manager.passNotificationToAI({ + .message = "first", + .actions = emptyActions(), + .priority = 5, + }); + manager.passNotificationToAI({ + .message = "second", + .actions = emptyActions(), + .priority = 5, + }); + manager.passNotificationToAI({ + .message = "third", + .actions = emptyActions(), + .priority = 5, + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } + + ASSERT_EQ(receivedMessages.size(), 3u); + EXPECT_THAT(receivedMessages[0], ::testing::StartsWith("first")); + EXPECT_THAT(receivedMessages[1], ::testing::StartsWith("second")); + EXPECT_THAT(receivedMessages[2], ::testing::StartsWith("third")); +} + +// ============================================================================ +// 5. Worker pin accumulation +// +// When a worker processes a pinned notification, the pin is added to the +// worker's pin set (the external set passed to run()). +// ============================================================================ +TEST_F(NotificationManagerUnitTest, PinAccumulation) { + NotificationManager manager; + ASet workerPins; + bool pinAdded = false; + + mAsync << manager.run(workerPins, [&](NotificationManager::Notification notification) -> AFuture { + // After processing, check that the pin was added. + if (notification.pin) { + pinAdded = true; + } + co_return false; // stop after one notification + }); + + // Pass a pinned notification. No worker has the pin yet, so pass an + // unpinned one first to wake the worker. + manager.passNotificationToAI({ + .message = "pinned msg", + .actions = emptyActions(), + .priority = 0, + .pin = "test-chat", + }); + manager.passNotificationToAI({ + .message = "wakeup", + .actions = emptyActions(), + .priority = 0, + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } + + // The worker should have taken the pinned notification (since no other + // worker owns the pin) and added it to the set. + EXPECT_TRUE(workerPins.contains("test-chat")); + EXPECT_TRUE(pinAdded); +} + +// ============================================================================ +// 6. Empty queue does not crash +// +// Calling passNotificationToAI and removeNotifications on an empty +// manager, or starting a worker with no notifications, should not crash. +// ============================================================================ +TEST_F(NotificationManagerUnitTest, EmptyQueue) { + NotificationManager manager; + + // Should not crash. + manager.removeNotifications("nothing"); + + // Should not crash. + { + ASet workerPins; + // Start and immediately stop a worker (pass an unpinned notification + // that tells the worker to stop). + mAsync << manager.run(workerPins, [&](NotificationManager::Notification) -> AFuture { + co_return false; + }); + + manager.passNotificationToAI({ + .message = "stop", + .actions = emptyActions(), + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } + } + + // Second passNotificationToAI after worker stopped. + manager.passNotificationToAI({ + .message = "after", + .actions = emptyActions(), + }); + // Ensure no crash when removing notifications that do/don't exist. + manager.removeNotifications("after"); + manager.removeNotifications("nonexistent"); +} + +// ============================================================================ +// 7. Pin stays with the worker that first processed it (2 workers) +// +// With two competing workers, a pinned notification that gets picked up by +// the first (idle) worker must keep being routed to that same worker for +// every subsequent notification carrying the same pin — even though the +// second worker is sitting idle the whole time and would normally be a +// candidate for unpinned notifications. +// ============================================================================ +TEST_F(NotificationManagerUnitTest, PinStaysWithWorkerThatFirstProcessedIt) { + NotificationManager manager; + ASet workerPinsA; + ASet workerPinsB; + // Give worker B an unrelated "control" pin so we can deterministically + // stop it later without affecting the "chat1" routing being tested. + workerPinsB << "control-b"; + + AVector receivedByA; + AVector receivedByB; + + // Register worker A first, then worker B - both start out idle. + mAsync << manager.run(workerPinsA, [&](NotificationManager::Notification notification) -> AFuture { + receivedByA << notification.message; + co_return !notification.message.contains("stopA"); + }); + mAsync << manager.run(workerPinsB, [&](NotificationManager::Notification notification) -> AFuture { + receivedByB << notification.message; + co_return !notification.message.contains("stopB"); + }); + + // 1. A pinned notification with no current owner is queued... + manager.passNotificationToAI({ + .message = "first", + .actions = emptyActions(), + .pin = "chat1", + }); + // 2. ...and an unpinned "kick" wakes the first idle worker (A), which + // drains the queue: it takes "first" (claiming the "chat1" pin) then + // "kick". + manager.passNotificationToAI({ + .message = "kick", + .actions = emptyActions(), + }); + + for (int i = 0; i < 1000 && receivedByA.size() < 2; ++i) { + mLoop.iteration(); + } + ASSERT_EQ(receivedByA.size(), 2u); + EXPECT_TRUE(workerPinsA.contains("chat1")); + + // 3. A second notification with the same pin must go straight to worker + // A, even though worker B has been idle this whole time. + manager.passNotificationToAI({ + .message = "second", + .actions = emptyActions(), + .pin = "chat1", + }); + + for (int i = 0; i < 1000 && receivedByA.size() < 3; ++i) { + mLoop.iteration(); + } + + ASSERT_EQ(receivedByA.size(), 3u); + EXPECT_THAT(receivedByA[0], ::testing::StartsWith("first")); + EXPECT_THAT(receivedByA[1], ::testing::StartsWith("kick")); + EXPECT_THAT(receivedByA[2], ::testing::StartsWith("second")); + // Worker B never saw any of the "chat1" traffic. + EXPECT_TRUE(receivedByB.empty()); + + // Cleanup: stop both workers deterministically via their owned pins. + manager.passNotificationToAI({ + .message = "stopA", + .actions = emptyActions(), + .pin = "chat1", + }); + manager.passNotificationToAI({ + .message = "stopB", + .actions = emptyActions(), + .pin = "control-b", + }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } +} + +// ============================================================================ +// 8. Two workers, each keeps its own pinned chat (isolation) +// +// When two workers each own a distinct pin, notifications for those pins +// must always be routed to their respective owner, never crossing over, +// regardless of send order or which worker happens to be idle. +// ============================================================================ +TEST_F(NotificationManagerUnitTest, TwoWorkersEachKeepOwnPinnedChat) { + NotificationManager manager; + ASet workerPinsA; + ASet workerPinsB; + workerPinsA << "chatA"; + workerPinsB << "chatB"; + + AVector receivedByA; + AVector receivedByB; + + mAsync << manager.run(workerPinsA, [&](NotificationManager::Notification notification) -> AFuture { + receivedByA << notification.message; + co_return !notification.message.contains("stopA"); + }); + mAsync << manager.run(workerPinsB, [&](NotificationManager::Notification notification) -> AFuture { + receivedByB << notification.message; + co_return !notification.message.contains("stopB"); + }); + + // Interleave notifications for both pinned chats. + manager.passNotificationToAI({ .message = "a1", .actions = emptyActions(), .pin = "chatA" }); + manager.passNotificationToAI({ .message = "b1", .actions = emptyActions(), .pin = "chatB" }); + manager.passNotificationToAI({ .message = "a2", .actions = emptyActions(), .pin = "chatA" }); + manager.passNotificationToAI({ .message = "b2", .actions = emptyActions(), .pin = "chatB" }); + + for (int i = 0; i < 1000 && (receivedByA.size() < 2 || receivedByB.size() < 2); ++i) { + mLoop.iteration(); + } + + ASSERT_EQ(receivedByA.size(), 2u); + ASSERT_EQ(receivedByB.size(), 2u); + EXPECT_THAT(receivedByA[0], ::testing::StartsWith("a1")); + EXPECT_THAT(receivedByA[1], ::testing::StartsWith("a2")); + EXPECT_THAT(receivedByB[0], ::testing::StartsWith("b1")); + EXPECT_THAT(receivedByB[1], ::testing::StartsWith("b2")); + + // Cleanup. + manager.passNotificationToAI({ .message = "stopA", .actions = emptyActions(), .pin = "chatA" }); + manager.passNotificationToAI({ .message = "stopB", .actions = emptyActions(), .pin = "chatB" }); + + while (!mAsync.empty()) { + mLoop.iteration(); + } +} \ No newline at end of file diff --git a/tests/PrometheusUnitTest.cpp b/tests/PrometheusUnitTest.cpp index 40f174c..1a7dac1 100644 --- a/tests/PrometheusUnitTest.cpp +++ b/tests/PrometheusUnitTest.cpp @@ -94,8 +94,8 @@ class PrometheusTestHarness : public AppBase { } protected: - void updateTools(OpenAITools& actions) override { - AppBase::updateTools(actions); + void updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext) override { + AppBase::updateTools(actions, temporaryContext); actions.insert({ .name = "my_tool", @@ -160,7 +160,7 @@ TEST_F(PrometheusUnitTest, ToolCallFiredOnSuccessfulToolCall) { SpyExporter spy; spy.registerAppBase(*app); - async << app->passNotificationToAI("Hello").onProcessed; + async << app->notificationManager().passNotificationToAI({"Hello"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -191,7 +191,7 @@ TEST_F(PrometheusUnitTest, ToolCallFiredCarriesCorrectToolName) { SpyExporter spy; spy.registerAppBase(*app); - async << app->passNotificationToAI("Hello").onProcessed; + async << app->notificationManager().passNotificationToAI({"Hello"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -219,8 +219,8 @@ TEST_F(PrometheusUnitTest, ToolCallFiredNotEmittedOnToolException) { public: explicit ThrowingHarness(_ ai) : PrometheusTestHarness(std::move(ai)) {} protected: - void updateTools(OpenAITools& actions) override { - PrometheusTestHarness::updateTools(actions); + void updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext) override { + PrometheusTestHarness::updateTools(actions, temporaryContext); // Insert a tool that throws actions.insert({ .name = "throwing_tool", @@ -237,7 +237,7 @@ TEST_F(PrometheusUnitTest, ToolCallFiredNotEmittedOnToolException) { SpyExporter spy; spy.registerAppBase(*app); - async << app->passNotificationToAI("Hello").onProcessed; + async << app->notificationManager().passNotificationToAI({"Hello"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -272,7 +272,7 @@ TEST_F(PrometheusUnitTest, ToolCallFiredBreadcrumbLabelsSnapshot) { SpyExporter spy; spy.registerAppBase(*app); - async << app->passNotificationToAI("Hello").onProcessed; + async << app->notificationManager().passNotificationToAI({"Hello"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -304,8 +304,8 @@ TEST_F(PrometheusUnitTest, ToolCallFiredlastOpenedChatLastMessageTime) { public: explicit TimedHarness(_ ai) : PrometheusTestHarness(std::move(ai)) {} protected: - void updateTools(OpenAITools& actions) override { - PrometheusTestHarness::updateTools(actions); + void updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext) override { + PrometheusTestHarness::updateTools(actions, temporaryContext); actions.insert({ .name = "my_tool", .description = "Test tool", @@ -323,7 +323,7 @@ TEST_F(PrometheusUnitTest, ToolCallFiredlastOpenedChatLastMessageTime) { SpyExporter spy; spy.registerAppBase(*app); - async << app->passNotificationToAI("Hello").onProcessed; + async << app->notificationManager().passNotificationToAI({"Hello"}).onProcessed; while (!async.empty()) { loop.iteration(); } @@ -362,12 +362,12 @@ TEST_F(PrometheusUnitTest, MultipleToolCallsAllCaptured) { SpyExporter spy; spy.registerAppBase(*app); - async << app->passNotificationToAI("First").onProcessed; + async << app->notificationManager().passNotificationToAI({"First"}).onProcessed; while (!async.empty()) { loop.iteration(); } - async << app->passNotificationToAI("Second").onProcessed; + async << app->notificationManager().passNotificationToAI({"Second"}).onProcessed; while (!async.empty()) { loop.iteration(); } diff --git a/tests/WebSearchIntegrationTest.cpp b/tests/WebSearchIntegrationTest.cpp index ee5ea81..40405f2 100644 --- a/tests/WebSearchIntegrationTest.cpp +++ b/tests/WebSearchIntegrationTest.cpp @@ -110,7 +110,7 @@ TEST(WebSearchIntegration, SearchAppAI) { IEventLoop::Handle h(&loop); AAsyncHolder async; - async << app->passNotificationToAI("You received a notification. Use #openChat").onProcessed; + async << app->notificationManager().passNotificationToAI({"You received a notification. Use #openChat"}).onProcessed; while (!async.empty()) { loop.iteration(); diff --git a/tests/common.h b/tests/common.h index 97de2cf..1699bc6 100644 --- a/tests/common.h +++ b/tests/common.h @@ -19,8 +19,8 @@ class AppMock : public AppBase { MOCK_METHOD(AString, openChat, (), ()); protected: - void updateTools(OpenAITools& actions) override { - AppBase::updateTools(actions); + void updateTools(OpenAITools& actions, const IOpenAIChat::Session& temporaryContext) override { + AppBase::updateTools(actions, temporaryContext); actions.insert({ .name = "send_telegram_message", .description = "Sends a message to the chat", diff --git a/tests/tools/ask_test.cpp b/tests/tools/ask_test.cpp index 6c95581..9c2c7e6 100644 --- a/tests/tools/ask_test.cpp +++ b/tests/tools/ask_test.cpp @@ -27,12 +27,6 @@ class DiaryMock : public Diary { .diaryDir = "/tmp/kuni_test_ask_diary", .openAI = nullptr, }) {} - - MOCK_METHOD( - (AFuture>), - query, - (const std::valarray& query, QueryOpts opts), - (override)); }; // --------------------------------------------------------------------------- @@ -119,6 +113,7 @@ TEST(AskTest, HandlerShortQueryError) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "short"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -142,6 +137,7 @@ TEST(AskTest, HandlerMissingQueryThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -175,9 +171,6 @@ TEST(AskTest, HandlerSuccessWithToolCall) { Diary::EntryExAndRelatedness hit{entryList.begin(), 0.9}; - EXPECT_CALL(diary, query(testing::_, testing::_)) - .WillOnce(Return(AFuture>(AVector{hit}))); - // LLM: first call returns #query tool call, second returns final answer EXPECT_CALL(*openAI, chatStreaming(testing::_, testing::_)) .WillOnce(Return(makeQueryToolCallResponse("What music does Alex write?"))) @@ -188,6 +181,7 @@ TEST(AskTest, HandlerSuccessWithToolCall) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "What kind of music does Alex write?"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -209,10 +203,6 @@ TEST(AskTest, HandlerLLMForcedToCallTool) { EXPECT_CALL(*openAI, embedding(testing::_, testing::_)) .WillOnce(Return(AFuture>(dummyEmbedding()))); - // diary returns no entries for simplicity - EXPECT_CALL(diary, query(testing::_, testing::_)) - .WillOnce(Return(AFuture>(AVector{}))); - // 1st call: no tool_calls (LLM skips step) → gets "you must perform at least one call" message // 2nd call: makes the #query tool call // 3rd call: returns final answer @@ -226,6 +216,7 @@ TEST(AskTest, HandlerLLMForcedToCallTool) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "Tell me about Alex's music habits in detail."}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -248,9 +239,6 @@ TEST(AskTest, HandlerWithTemporaryContextEnrichesQuery) { EXPECT_CALL(*openAI, embedding(testing::_, testing::_)) .WillOnce(Return(AFuture>(dummyEmbedding()))); - EXPECT_CALL(diary, query(testing::_, testing::_)) - .WillOnce(Return(AFuture>(AVector{}))); - // Capture the messages passed to chat to verify query enrichment AString capturedUserContent; EXPECT_CALL(*openAI, chatStreaming(testing::_, testing::_)) @@ -271,6 +259,7 @@ TEST(AskTest, HandlerWithTemporaryContextEnrichesQuery) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "Does Alex play any musical instruments?"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -293,10 +282,6 @@ TEST(AskTest, HandlerDiaryReturnsNoEntries) { EXPECT_CALL(*openAI, embedding(testing::_, testing::_)) .WillOnce(Return(AFuture>(dummyEmbedding()))); - // Empty result from diary - EXPECT_CALL(diary, query(testing::_, testing::_)) - .WillOnce(Return(AFuture>(AVector{}))); - EXPECT_CALL(*openAI, chatStreaming(testing::_, testing::_)) .WillOnce(Return(makeQueryToolCallResponse("user hobbies"))) .WillOnce(Return(makeFinalResponse(kFinalAnswer))); @@ -306,6 +291,7 @@ TEST(AskTest, HandlerDiaryReturnsNoEntries) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "What are the user's hobbies and interests?"}}, + .temporaryContext = {}, .allToolCalls = {}, })); diff --git a/tests/tools/get_chat_photo_test.cpp b/tests/tools/get_chat_photo_test.cpp index b24c63a..b014949 100644 --- a/tests/tools/get_chat_photo_test.cpp +++ b/tests/tools/get_chat_photo_test.cpp @@ -91,6 +91,7 @@ TEST(GetChatPhotoTest, NoPhoto) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -153,6 +154,7 @@ TEST(GetChatPhotoTest, HasPhotoSuccess) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, + .temporaryContext = {}, .allToolCalls = {}, })); diff --git a/tests/tools/react_with_emoji_test.cpp b/tests/tools/react_with_emoji_test.cpp index ff80d09..2b57d46 100644 --- a/tests/tools/react_with_emoji_test.cpp +++ b/tests/tools/react_with_emoji_test.cpp @@ -72,6 +72,7 @@ TEST(ReactWithEmojiTest, Success) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"message_id", 42}, {"emoji", "🔥"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -95,6 +96,7 @@ TEST(ReactWithEmojiTest, WrongChatId) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", 99999}, {"message_id", 42}, {"emoji", "🔥"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -119,6 +121,7 @@ TEST(ReactWithEmojiTest, MissingMessageIdThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"emoji", "🔥"}}, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -141,6 +144,7 @@ TEST(ReactWithEmojiTest, MissingEmojiThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"message_id", 42}}, + .temporaryContext = {}, .allToolCalls = {}, })), AException diff --git a/tests/tools/remove_and_ban_chat_test.cpp b/tests/tools/remove_and_ban_chat_test.cpp index 43b4079..5c05767 100644 --- a/tests/tools/remove_and_ban_chat_test.cpp +++ b/tests/tools/remove_and_ban_chat_test.cpp @@ -46,6 +46,7 @@ TEST(RemoveAndBanChatTest, PapikChatIdReturnsFailed) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", config().papikChatId}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -87,6 +88,7 @@ TEST(RemoveAndBanChatTest, BasicGroupCallsLeaveChat) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", CHAT_ID}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -126,6 +128,7 @@ TEST(RemoveAndBanChatTest, SupergroupCallsLeaveChat) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", CHAT_ID}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -147,6 +150,7 @@ TEST(RemoveAndBanChatTest, MissingChatIdThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, // empty args, no "chat_id" + .temporaryContext = {}, .allToolCalls = {}, })), AException diff --git a/tests/tools/search_chats_test.cpp b/tests/tools/search_chats_test.cpp index d8ee191..cb365c4 100644 --- a/tests/tools/search_chats_test.cpp +++ b/tests/tools/search_chats_test.cpp @@ -46,6 +46,7 @@ TEST(SearchChatsTest, MissingQueryThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, // empty args, no "query" + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -83,6 +84,7 @@ TEST(SearchChatsTest, NoResults) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "nonexistent_chat"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -134,6 +136,7 @@ TEST(SearchChatsTest, WithResults) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "test_chat"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -176,6 +179,7 @@ TEST(SearchChatsTest, AtPrefixStripped) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"query", "@username"}}, + .temporaryContext = {}, .allToolCalls = {}, })); diff --git a/tests/tools/send_telegram_message_test.cpp b/tests/tools/send_telegram_message_test.cpp index 674e7da..9e2433a 100644 --- a/tests/tools/send_telegram_message_test.cpp +++ b/tests/tools/send_telegram_message_test.cpp @@ -124,6 +124,7 @@ TEST(SendTelegramMessageTest, SuccessSimpleText) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"text", "Hello!"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -157,6 +158,7 @@ TEST(SendTelegramMessageTest, WrongChatId) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"text", "Hello!"}, {"chat_id", 99999}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -192,6 +194,7 @@ TEST(SendTelegramMessageTest, MissingAllContent) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, // empty args + .temporaryContext = {}, .allToolCalls = {}, })); @@ -231,6 +234,7 @@ TEST(SendTelegramMessageTest, BothPhotoAndAudioError) { {"photo_filename", "photo.jpg"}, {"audio_filename", "audio.ogg"}, }, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -272,6 +276,7 @@ TEST(SendTelegramMessageTest, ReplyToMessageFromAnotherChatThrows) { {"text", "Hello!"}, {"reply_to_message_id", 42}, }, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -310,6 +315,7 @@ TEST(SendTelegramMessageTest, ReplyToExistingMessage) { {"text", "Hello!"}, {"reply_to_message_id", 42}, }, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -346,6 +352,7 @@ TEST(SendTelegramMessageTest, TooManyMessagesInRowThrows) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"text", "msg{}"_format(i)}}, + .temporaryContext = {}, .allToolCalls = {}, })); EXPECT_TRUE(result.contains("sent successfully")) << "at iteration " << i << ": " << result; @@ -384,6 +391,7 @@ TEST(SendTelegramMessageTest, MultiLineMessageGetsSplit) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"text", "line1\nline2\nline3"}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -420,6 +428,7 @@ TEST(SendTelegramMessageTest, InvalidPhotoFilenameSlash) { {"text", "hello"}, {"photo_filename", "subdir/photo.jpg"}, }, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -456,6 +465,7 @@ TEST(SendTelegramMessageTest, InvalidPhotoFilenameDotDot) { {"text", "hello"}, {"photo_filename", "../photo.jpg"}, }, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -491,6 +501,7 @@ TEST(SendTelegramMessageTest, InvalidAudioFilenameSlash) { .args = AJson::Object{ {"audio_filename", "subdir/audio.ogg"}, }, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -525,6 +536,7 @@ TEST(SendTelegramMessageTest, InvalidAudioFilenameDotDot) { .args = AJson::Object{ {"audio_filename", "../audio.ogg"}, }, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -570,6 +582,7 @@ TEST(SendTelegramMessageTest, RepeatDetectionThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"text", "Hello there!"}}, + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -603,6 +616,7 @@ TEST(SendTelegramMessageTest, FirstMessageEncouragesFollowUp) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"text", "Hi!"}}, + .temporaryContext = {}, .allToolCalls = {}, })); diff --git a/tests/tools/take_photo_test.cpp b/tests/tools/take_photo_test.cpp index f7243f4..9831460 100644 --- a/tests/tools/take_photo_test.cpp +++ b/tests/tools/take_photo_test.cpp @@ -39,6 +39,7 @@ TEST(TakePhotoTest, MissingPhotoDescThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{}, // empty args, no "photo_desc" + .temporaryContext = {}, .allToolCalls = {}, })), AException @@ -62,6 +63,7 @@ TEST(TakePhotoTest, PhotoDescNotStringThrows) { util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"photo_desc", 123}}, + .temporaryContext = {}, .allToolCalls = {}, })), AException diff --git a/tests/tools/view_messages_around_test.cpp b/tests/tools/view_messages_around_test.cpp index e6504a6..dc6abcd 100644 --- a/tests/tools/view_messages_around_test.cpp +++ b/tests/tools/view_messages_around_test.cpp @@ -62,6 +62,7 @@ TEST(ViewMessagesAroundTest, LockdownBlocksInaccessibleChat) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", config().papikChatId + 1}, {"message_id", 1}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -90,6 +91,7 @@ TEST(ViewMessagesAroundTest, MessageNotFound) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", chatId}, {"message_id", 42}}, + .temporaryContext = {}, .allToolCalls = {}, })); @@ -141,6 +143,7 @@ TEST(ViewMessagesAroundTest, ReturnsSurroundingMessages) { auto result = util::await_synchronously(tool.handler({ .tools = tools, .args = AJson::Object{{"chat_id", chatId}, {"message_id", targetId}, {"before", 1}, {"after", 1}}, + .temporaryContext = {}, .allToolCalls = {}, }));