From c969703319222d9ee57fc9a722277a362ba63bac Mon Sep 17 00:00:00 2001 From: amoya Date: Mon, 21 Sep 2026 13:57:15 +0200 Subject: [PATCH] [ADD] ai_connection_queue --- ai_connection/models/ai_connection.py | 143 ++++-- ai_connection_queue/README.rst | 103 ++++ ai_connection_queue/__init__.py | 1 + ai_connection_queue/__manifest__.py | 19 + ai_connection_queue/models/__init__.py | 2 + ai_connection_queue/models/ai_connection.py | 164 +++++++ .../models/ai_connection_run.py | 72 +++ ai_connection_queue/pyproject.toml | 3 + ai_connection_queue/readme/CONTEXT.md | 10 + ai_connection_queue/readme/CONTRIBUTORS.md | 2 + ai_connection_queue/readme/DESCRIPTION.md | 7 + .../security/ir.model.access.csv | 3 + .../static/description/index.html | 450 ++++++++++++++++++ ai_connection_queue/tests/__init__.py | 1 + ai_connection_queue/tests/fake_models.py | 26 + .../tests/test_ai_connection_queue.py | 169 +++++++ 16 files changed, 1131 insertions(+), 44 deletions(-) create mode 100644 ai_connection_queue/README.rst create mode 100644 ai_connection_queue/__init__.py create mode 100644 ai_connection_queue/__manifest__.py create mode 100644 ai_connection_queue/models/__init__.py create mode 100644 ai_connection_queue/models/ai_connection.py create mode 100644 ai_connection_queue/models/ai_connection_run.py create mode 100644 ai_connection_queue/pyproject.toml create mode 100644 ai_connection_queue/readme/CONTEXT.md create mode 100644 ai_connection_queue/readme/CONTRIBUTORS.md create mode 100644 ai_connection_queue/readme/DESCRIPTION.md create mode 100644 ai_connection_queue/security/ir.model.access.csv create mode 100644 ai_connection_queue/static/description/index.html create mode 100644 ai_connection_queue/tests/__init__.py create mode 100644 ai_connection_queue/tests/fake_models.py create mode 100644 ai_connection_queue/tests/test_ai_connection_queue.py diff --git a/ai_connection/models/ai_connection.py b/ai_connection/models/ai_connection.py index 3c129d45..704c055a 100644 --- a/ai_connection/models/ai_connection.py +++ b/ai_connection/models/ai_connection.py @@ -51,56 +51,111 @@ def _run( ) def _run_ai(self, messages, tools=None, record=None, max_iterations=None): - client = getattr(self, f"_get_client_{self.kind}")(tools) - # Shallow copying messages to avoid edition of the messages - messages = list(messages) + self.ensure_one() if max_iterations is None: max_iterations = self._max_iterations - iteration = 0 - prompt_tokens = 0 - completion_tokens = 0 - while iteration < max_iterations: - iteration += 1 - response = client.handle_message( - messages=messages, temperature=self.temperature - ) - messages.append(response["message"]) - prompt_tokens += response.get("usage", {}).get("prompt_tokens", 0) - completion_tokens += response.get("usage", {}).get("completion_tokens", 0) - if not response.get("tool_calls"): - return ( - response["message"]["content"], - prompt_tokens, - completion_tokens, - iteration, - ) - for tool_call in response["tool_calls"]: - tool = tools.filtered( - lambda t, tool_call=tool_call: t.name == tool_call["name"] + return self._run_ai_step( + messages=list(messages), + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=0, + prompt_tokens=0, + completion_tokens=0, + ) + + def _run_ai_step( + self, + messages, + tools, + record, + max_iterations, + iteration, + prompt_tokens, + completion_tokens, + ): + self.ensure_one() + if iteration >= max_iterations: + return self._run_ai_error( + self.env._( + "Iterations reached the maximum allowed (%s)", max_iterations ) - if tool: - try: - with self.env.cr.savepoint(): - messages.append( - self._process_tool_call(tool, tool_call, record) - ) - except Exception as e: - getattr( - self, - f"_process_tool_call_result_{self.kind}", - self._process_tool_call_result, - )( - tool, - { - "error": str(e), - "type": type(e).__name__, - }, - tool_call, + ) + iteration += 1 + + client = getattr(self, f"_get_client_{self.kind}")(tools) + response = client.handle_message( + messages=messages, temperature=self.temperature + ) + messages = [*messages, response["message"]] + prompt_tokens += response.get("usage", {}).get("prompt_tokens", 0) + completion_tokens += response.get("usage", {}).get("completion_tokens", 0) + + if not response.get("tool_calls"): + return self._run_ai_finalize( + messages, prompt_tokens, completion_tokens, iteration + ) + + for tool_call in response["tool_calls"]: + tool = tools.filtered( + lambda t, tool_call=tool_call: t.name == tool_call["name"] + ) + if tool: + try: + with self.env.cr.savepoint(): + messages.append( + self._process_tool_call(tool, tool_call, record) ) - raise UserError( - self.env._("Iterations reached the maximum allowed (%s)", max_iterations) + except Exception as e: + getattr( + self, + f"_process_tool_call_result_{self.kind}", + self._process_tool_call_result, + )( + tool, + { + "error": str(e), + "type": type(e).__name__, + }, + tool_call, + ) + + return self._run_ai_next( + messages=messages, + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=iteration, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, ) + def _run_ai_next( + self, + messages, + tools, + record, + max_iterations, + iteration, + prompt_tokens, + completion_tokens, + ): + return self._run_ai_step( + messages=messages, + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=iteration, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + + def _run_ai_finalize(self, messages, prompt_tokens, completion_tokens, iteration): + return (messages[-1]["content"], prompt_tokens, completion_tokens, iteration) + + def _run_ai_error(self, message): + raise UserError(message) + def _process_tool_call(self, tool, tool_call, record): tool_response = tool._execute_tool(**tool_call["arguments"], record=record) return getattr( diff --git a/ai_connection_queue/README.rst b/ai_connection_queue/README.rst new file mode 100644 index 00000000..ff87547d --- /dev/null +++ b/ai_connection_queue/README.rst @@ -0,0 +1,103 @@ +.. image:: https://odoo-community.org/readme-banner-image + :target: https://odoo-community.org/get-involved?utm_source=readme + :alt: Odoo Community Association + +=================== +Ai Connection Queue +=================== + +.. + !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + !! This file is generated by oca-gen-addon-readme !! + !! changes will be overwritten. !! + !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + !! source digest: sha256:0a4dbc891f9e686adb7d09126a55a49ae5c9f8e373de773c908ab50f0c41c273 + !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + +.. |badge1| image:: https://img.shields.io/badge/maturity-Beta-yellow.png + :target: https://odoo-community.org/page/development-status + :alt: Beta +.. |badge2| image:: https://img.shields.io/badge/license-AGPL--3-blue.png + :target: http://www.gnu.org/licenses/agpl-3.0-standalone.html + :alt: License: AGPL-3 +.. |badge3| image:: https://img.shields.io/badge/github-OCA%2Fai-lightgray.png?logo=github + :target: https://github.com/OCA/ai/tree/18.0/ai_connection_queue + :alt: OCA/ai +.. |badge4| image:: https://img.shields.io/badge/weblate-Translate%20me-F47D42.png + :target: https://translation.odoo-community.org/projects/ai-18-0/ai-18-0-ai_connection_queue + :alt: Translate me on Weblate +.. |badge5| image:: https://img.shields.io/badge/runboat-Try%20me-875A7B.png + :target: https://runboat.odoo-community.org/builds?repo=OCA/ai&target_branch=18.0 + :alt: Try me on Runboat + +|badge1| |badge2| |badge3| |badge4| |badge5| + +This module runs ``ai.connection`` conversations through ``queue_job`` +instead of blocking the calling worker until the AI provider produces a +final answer. + +It does not change ``ai.connection``'s synchronous behaviour at all - it +only adds an opt-in, context-driven way to run a conversation +asynchronously and be notified of its progress. + +**Table of contents** + +.. contents:: + :local: + +Use Cases / Context +=================== + +This module runs ``ai.connection`` conversations through ``queue_job``, +one job per round, instead of blocking the calling worker until a final +answer comes back. + +It is enabled per call through context, without changing +``ai.connection``'s own methods, and it can also notify a caller-chosen +record or model after every round, not only when the conversation +finishes. + +Check ``ai.connection.run`` to follow the state of an asynchronous +conversation. + +Bug Tracker +=========== + +Bugs are tracked on `GitHub Issues `_. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +`feedback `_. + +Do not contact contributors directly about support or help with technical issues. + +Credits +======= + +Authors +------- + +* SDi + +Contributors +------------ + +- `SDi `__ + + - Angel Moya + +Maintainers +----------- + +This module is maintained by the OCA. + +.. image:: https://odoo-community.org/logo.png + :alt: Odoo Community Association + :target: https://odoo-community.org + +OCA, or the Odoo Community Association, is a nonprofit organization whose +mission is to support the collaborative development of Odoo features and +promote its widespread use. + +This module is part of the `OCA/ai `_ project on GitHub. + +You are welcome to contribute. To learn how please visit https://odoo-community.org/page/Contribute. diff --git a/ai_connection_queue/__init__.py b/ai_connection_queue/__init__.py new file mode 100644 index 00000000..0650744f --- /dev/null +++ b/ai_connection_queue/__init__.py @@ -0,0 +1 @@ +from . import models diff --git a/ai_connection_queue/__manifest__.py b/ai_connection_queue/__manifest__.py new file mode 100644 index 00000000..1891e23a --- /dev/null +++ b/ai_connection_queue/__manifest__.py @@ -0,0 +1,19 @@ +# Copyright 2026 SDi +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +{ + "name": "Ai Connection Queue", + "summary": """Run ai.connection conversations through queue_job""", + "version": "18.0.1.0.0", + "license": "AGPL-3", + "author": "SDi,Odoo Community Association (OCA)", + "website": "https://github.com/OCA/ai", + "depends": [ + "ai_connection", + "queue_job", + ], + "data": [ + "security/ir.model.access.csv", + ], + "demo": [], +} diff --git a/ai_connection_queue/models/__init__.py b/ai_connection_queue/models/__init__.py new file mode 100644 index 00000000..92813f9e --- /dev/null +++ b/ai_connection_queue/models/__init__.py @@ -0,0 +1,2 @@ +from . import ai_connection_run +from . import ai_connection diff --git a/ai_connection_queue/models/ai_connection.py b/ai_connection_queue/models/ai_connection.py new file mode 100644 index 00000000..c5a97c26 --- /dev/null +++ b/ai_connection_queue/models/ai_connection.py @@ -0,0 +1,164 @@ +# Copyright 2026 SDi +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo import models + +ASYNC_CONTEXT_KEY = "ai_connection_async" +NOTIFY_MODEL_CONTEXT_KEY = "ai_connection_notify_model" +NOTIFY_RES_ID_CONTEXT_KEY = "ai_connection_notify_res_id" +NOTIFY_METHOD_CONTEXT_KEY = "ai_connection_notify_method" +RUN_CONTEXT_KEY = "ai_connection_run_id" + + +class AiConnection(models.Model): + _inherit = "ai.connection" + + def _job_prepare_context_before_enqueue_keys(self): + return super()._job_prepare_context_before_enqueue_keys() + (RUN_CONTEXT_KEY,) + + def _run_ai_step( + self, + messages, + tools, + record, + max_iterations, + iteration, + prompt_tokens, + completion_tokens, + ): + if iteration == 0 and RUN_CONTEXT_KEY not in self.env.context: + if self.env.context.get(ASYNC_CONTEXT_KEY): + run = self._create_async_run(messages, tools, record, max_iterations) + self._dispatch_round( + run, + messages=messages, + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=iteration, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + return run.id + + return super()._run_ai_step( + messages=messages, + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=iteration, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + + def _create_async_run(self, messages, tools, record, max_iterations): + self.ensure_one() + notify_model = self.env.context.get(NOTIFY_MODEL_CONTEXT_KEY) + notify_res_id = self.env.context.get(NOTIFY_RES_ID_CONTEXT_KEY) + notify_method = self.env.context.get(NOTIFY_METHOD_CONTEXT_KEY) + if bool(notify_model) != bool(notify_method): + raise ValueError( + f"{NOTIFY_MODEL_CONTEXT_KEY} and {NOTIFY_METHOD_CONTEXT_KEY} " + "must be given together." + ) + return self.env["ai.connection.run"].create( + { + "connection_id": self.id, + "messages": list(messages), + "tool_ids": [(6, 0, tools.ids)] if tools else False, + "res_model": record._name if record else False, + "res_id": record.id if record else False, + "max_iterations": max_iterations or self._max_iterations, + "notify_res_model": notify_model or False, + "notify_res_id": notify_res_id or False, + "notify_method": notify_method or False, + } + ) + + def _dispatch_round(self, run, **kwargs): + """Queue one round for `run`, tagging it with `run`'s id via context + so it (and everything it recurses into) can find its way back to it. + + A distinct `identity_key` per round matters here: without it, a + retried or duplicated dispatch for the same round could run twice in + parallel and corrupt `run`'s state (two rounds both reading + iteration N and writing iteration N+1 independently). + """ + self.with_context(**{RUN_CONTEXT_KEY: run.id}).with_delay( + identity_key=f"ai-connection-run-{run.id}-{kwargs['iteration']}" + )._run_ai_step(**kwargs) + + def _run_ai_next( + self, + messages, + tools, + record, + max_iterations, + iteration, + prompt_tokens, + completion_tokens, + ): + run_id = self.env.context.get(RUN_CONTEXT_KEY) + if not run_id: + return super()._run_ai_next( + messages=messages, + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=iteration, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + + run = self.env["ai.connection.run"].browse(run_id) + run.write( + { + "messages": messages, + "iteration": iteration, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + } + ) + run._notify_progress() + self._dispatch_round( + run, + messages=messages, + tools=tools, + record=record, + max_iterations=max_iterations, + iteration=iteration, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + return None + + def _run_ai_finalize(self, messages, prompt_tokens, completion_tokens, iteration): + run_id = self.env.context.get(RUN_CONTEXT_KEY) + if not run_id: + return super()._run_ai_finalize( + messages, prompt_tokens, completion_tokens, iteration + ) + + run = self.env["ai.connection.run"].browse(run_id) + run.write( + { + "state": "done", + "result": messages[-1]["content"], + "messages": messages, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "iteration": iteration, + } + ) + run._notify_progress() + return None + + def _run_ai_error(self, message): + run_id = self.env.context.get(RUN_CONTEXT_KEY) + if not run_id: + return super()._run_ai_error(message) + + run = self.env["ai.connection.run"].browse(run_id) + run.write({"state": "error", "error_message": message}) + run._notify_progress() + return None diff --git a/ai_connection_queue/models/ai_connection_run.py b/ai_connection_queue/models/ai_connection_run.py new file mode 100644 index 00000000..90cf4384 --- /dev/null +++ b/ai_connection_queue/models/ai_connection_run.py @@ -0,0 +1,72 @@ +# Copyright 2026 SDi +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo import fields, models +from odoo.exceptions import UserError + + +class AiConnectionRun(models.Model): + _name = "ai.connection.run" + _description = "AI Connection Run" + _order = "id desc" + + connection_id = fields.Many2one( + "ai.connection", required=True, ondelete="cascade", readonly=True + ) + state = fields.Selection( + [ + ("running", "Running"), + ("done", "Done"), + ("error", "Error"), + ], + default="running", + required=True, + readonly=True, + ) + + messages = fields.Json(default=list, readonly=True) + tool_ids = fields.Many2many("ai.tool", readonly=True) + res_model = fields.Char(readonly=True) + res_id = fields.Integer(readonly=True) + max_iterations = fields.Integer(readonly=True) + iteration = fields.Integer(default=0, readonly=True) + prompt_tokens = fields.Integer(default=0, readonly=True) + completion_tokens = fields.Integer(default=0, readonly=True) + result = fields.Text(readonly=True) + error_message = fields.Text(readonly=True) + notify_res_model = fields.Char(readonly=True) + notify_res_id = fields.Integer(readonly=True) + notify_method = fields.Char(readonly=True) + + def _get_record(self): + self.ensure_one() + if not self.res_model or not self.res_id: + return None + return self.env[self.res_model].browse(self.res_id) + + def _get_notify_target(self): + self.ensure_one() + if not (self.notify_res_model and self.notify_method): + return None + if not self.notify_res_id: + return self.env[self.notify_res_model] + target = self.env[self.notify_res_model].browse(self.notify_res_id) + return target if target.exists() else None + + def is_done(self): + self.ensure_one() + return self.state != "running" + + def get_result(self): + self.ensure_one() + if self.state == "running": + raise UserError(self.env._("This AI connection run is not finished yet.")) + if self.state == "error": + raise UserError(self.error_message) + return (self.result, self.prompt_tokens, self.completion_tokens, self.iteration) + + def _notify_progress(self): + self.ensure_one() + target = self._get_notify_target() + if target is not None: + getattr(target, self.notify_method)(self) diff --git a/ai_connection_queue/pyproject.toml b/ai_connection_queue/pyproject.toml new file mode 100644 index 00000000..4231d0cc --- /dev/null +++ b/ai_connection_queue/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["whool"] +build-backend = "whool.buildapi" diff --git a/ai_connection_queue/readme/CONTEXT.md b/ai_connection_queue/readme/CONTEXT.md new file mode 100644 index 00000000..0fbd8dbc --- /dev/null +++ b/ai_connection_queue/readme/CONTEXT.md @@ -0,0 +1,10 @@ +This module runs `ai.connection` conversations through `queue_job`, one job +per round, instead of blocking the calling worker until a final answer +comes back. + +It is enabled per call through context, without changing `ai.connection`'s +own methods, and it can also notify a caller-chosen record or model after +every round, not only when the conversation finishes. + +Check `ai.connection.run` to follow the state of an asynchronous +conversation. diff --git a/ai_connection_queue/readme/CONTRIBUTORS.md b/ai_connection_queue/readme/CONTRIBUTORS.md new file mode 100644 index 00000000..56ae9751 --- /dev/null +++ b/ai_connection_queue/readme/CONTRIBUTORS.md @@ -0,0 +1,2 @@ +- [SDi](https://sdi.es) + - Angel Moya diff --git a/ai_connection_queue/readme/DESCRIPTION.md b/ai_connection_queue/readme/DESCRIPTION.md new file mode 100644 index 00000000..9c2c947a --- /dev/null +++ b/ai_connection_queue/readme/DESCRIPTION.md @@ -0,0 +1,7 @@ +This module runs `ai.connection` conversations through `queue_job` instead +of blocking the calling worker until the AI provider produces a final +answer. + +It does not change `ai.connection`'s synchronous behaviour at all - it only +adds an opt-in, context-driven way to run a conversation asynchronously and +be notified of its progress. diff --git a/ai_connection_queue/security/ir.model.access.csv b/ai_connection_queue/security/ir.model.access.csv new file mode 100644 index 00000000..8c3bbc20 --- /dev/null +++ b/ai_connection_queue/security/ir.model.access.csv @@ -0,0 +1,3 @@ +id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink +access_ai_connection_run,access_ai_connection_run,model_ai_connection_run,base.group_user,1,0,0,0 +manage_ai_connection_run,manage_ai_connection_run,model_ai_connection_run,base.group_system,1,1,1,0 diff --git a/ai_connection_queue/static/description/index.html b/ai_connection_queue/static/description/index.html new file mode 100644 index 00000000..d763a51c --- /dev/null +++ b/ai_connection_queue/static/description/index.html @@ -0,0 +1,450 @@ + + + + + +Ai Connection Queue + + + +
+ + + +Odoo Community Association + +
+

Ai Connection Queue

+ +

Beta License: AGPL-3 OCA/ai Translate me on Weblate Try me on Runboat

+

This module runs ai.connection conversations through queue_job +instead of blocking the calling worker until the AI provider produces a +final answer.

+

It does not change ai.connection’s synchronous behaviour at all - it +only adds an opt-in, context-driven way to run a conversation +asynchronously and be notified of its progress.

+

Table of contents

+ +
+

Use Cases / Context

+

This module runs ai.connection conversations through queue_job, +one job per round, instead of blocking the calling worker until a final +answer comes back.

+

It is enabled per call through context, without changing +ai.connection’s own methods, and it can also notify a caller-chosen +record or model after every round, not only when the conversation +finishes.

+

Check ai.connection.run to follow the state of an asynchronous +conversation.

+
+
+

Bug Tracker

+

Bugs are tracked on GitHub Issues. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +feedback.

+

Do not contact contributors directly about support or help with technical issues.

+
+
+

Credits

+
+

Authors

+
    +
  • SDi
    • +
+
+
+

Contributors

+
    +
  • SDi
      +
    • Angel Moya
      • +
    +
    • +
+
+
+

Maintainers

+

This module is maintained by the OCA.

+ +Odoo Community Association + +

OCA, or the Odoo Community Association, is a nonprofit organization whose +mission is to support the collaborative development of Odoo features and +promote its widespread use.

+

This module is part of the OCA/ai project on GitHub.

+

You are welcome to contribute. To learn how please visit https://odoo-community.org/page/Contribute.

+
+
+
+
+ + diff --git a/ai_connection_queue/tests/__init__.py b/ai_connection_queue/tests/__init__.py new file mode 100644 index 00000000..8d9e86f5 --- /dev/null +++ b/ai_connection_queue/tests/__init__.py @@ -0,0 +1 @@ +from . import test_ai_connection_queue diff --git a/ai_connection_queue/tests/fake_models.py b/ai_connection_queue/tests/fake_models.py new file mode 100644 index 00000000..dfe19018 --- /dev/null +++ b/ai_connection_queue/tests/fake_models.py @@ -0,0 +1,26 @@ +# Copyright 2026 SDi +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo import api, models + +# A plain in-process list is enough here: tests drive every queued round +# synchronously through trap_jobs(), so this always runs in the same Python +# process as the assertions reading it - a real, cross-process run would +# have to observe this through the `ai.connection.run` record itself, not a +# module-level variable like this one. +notify_calls = [] +notify_model_calls = [] + + +class ResPartnerNotifyReceiver(models.Model): + _inherit = "res.partner" + + def _ai_connection_run_test_callback(self, run): + notify_calls.append((self.id, run.state, run.iteration)) + + @api.model + def _ai_connection_run_test_model_callback(self, run): + # `self` is whatever recordset the callback was called on - a + # model-level (notify_model) call resolves it as env[model], i.e. + # empty, same as any other `@api.model` call. + notify_model_calls.append((tuple(self.ids), run.state, run.iteration)) diff --git a/ai_connection_queue/tests/test_ai_connection_queue.py b/ai_connection_queue/tests/test_ai_connection_queue.py new file mode 100644 index 00000000..3cd16e92 --- /dev/null +++ b/ai_connection_queue/tests/test_ai_connection_queue.py @@ -0,0 +1,169 @@ +# Copyright 2026 SDi +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo_test_helper import FakeModelLoader + +from odoo.exceptions import UserError +from odoo.tests.common import TransactionCase + +from odoo.addons.queue_job.tests.common import JobMixin + + +class TestAiConnectionQueue(JobMixin, TransactionCase): + def setUp(self): + super().setUp() + self.loader = FakeModelLoader(self.env, self.__module__) + self.loader.backup_registry() + # Reuse ai_connection's own demo provider (kind="demo") instead of + # duplicating a fake AI client here. + from odoo.addons.ai_connection.tests.fake_models import AiConnection + + from .fake_models import ResPartnerNotifyReceiver + + self.loader.update_registry((AiConnection, ResPartnerNotifyReceiver)) + self.addCleanup(self.loader.restore_registry) + + def _run_all_rounds(self, trap): + """Drive every round the `ai_connection_async` context key enqueues, + one at a time - each round's execution can itself enqueue the next + one, so a single `perform_enqueued_jobs()` is not enough in general. + """ + while trap.enqueued_jobs: + trap.perform_enqueued_jobs() + + def test_async_run_returns_id_and_completes(self): + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.trap_jobs() as trap: + run_id = connection.with_context(ai_connection_async=True)._run( + "Hello, AI!" + ) + # The caller gets an id back immediately, not a recordset and + # not an answer: at this point the first round has only been + # enqueued, not executed. + self.assertIsInstance(run_id, int) + run = self.env["ai.connection.run"].browse(run_id) + self.assertEqual(run.state, "running") + self.assertFalse(run.is_done()) + + trap.assert_jobs_count(1) + self._run_all_rounds(trap) + + self.assertTrue(run.is_done()) + self.assertEqual(run.state, "done") + content, prompt_tokens, completion_tokens, iteration = run.get_result() + self.assertEqual(content, "This is a demo response to the prompt: Hello, AI!") + self.assertEqual(iteration, 1) + + def test_async_run_with_tool_spans_multiple_rounds(self): + tool = self.env.ref("ai_tool.current_date") + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.trap_jobs() as trap: + run_id = connection.with_context(ai_connection_async=True)._run( + "get_date", tools=tool + ) + run = self.env["ai.connection.run"].browse(run_id) + self._run_all_rounds(trap) + + self.assertEqual(run.state, "done") + self.assertEqual(run.iteration, 2) + self.assertIn("get_date", run.tool_ids.mapped("name")) + + def test_async_run_max_iterations_reports_error_without_raising(self): + tool = self.env.ref("ai_tool.current_date") + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.trap_jobs() as trap: + # Never raises here: by the time max_iterations would be hit, + # this call has long since returned an id. + run_id = connection.with_context(ai_connection_async=True)._run( + "get_date", tools=tool, max_iterations=1 + ) + run = self.env["ai.connection.run"].browse(run_id) + self._run_all_rounds(trap) + + self.assertEqual(run.state, "error") + with self.assertRaises(UserError): + run.get_result() + + def test_async_run_notifies_callback_on_every_round(self): + from .fake_models import notify_calls + + notify_calls.clear() + partner = self.env.user.partner_id + tool = self.env.ref("ai_tool.current_date") + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.trap_jobs() as trap: + run_id = connection.with_context( + ai_connection_async=True, + ai_connection_notify_model=partner._name, + ai_connection_notify_res_id=partner.id, + ai_connection_notify_method="_ai_connection_run_test_callback", + )._run("get_date", tools=tool) + self._run_all_rounds(trap) + + run = self.env["ai.connection.run"].browse(run_id) + self.assertEqual(run.state, "done") + # One callback per round (this scenario takes 2: the tool call, then + # the final answer) - not only when the run finishes. + self.assertEqual( + notify_calls, [(partner.id, "running", 1), (partner.id, "done", 2)] + ) + + def test_async_run_notifies_model_level_callback(self): + from .fake_models import notify_model_calls + + notify_model_calls.clear() + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.trap_jobs() as trap: + run_id = connection.with_context( + ai_connection_async=True, + ai_connection_notify_model="res.partner", + # ai_connection_notify_res_id deliberately omitted. + ai_connection_notify_method="_ai_connection_run_test_model_callback", + )._run("Hello, AI!") + self._run_all_rounds(trap) + + run = self.env["ai.connection.run"].browse(run_id) + self.assertEqual(run.state, "done") + # No record ids: this is the @api.model-style, env[model] call, not + # bound to any particular res.partner. + self.assertEqual(notify_model_calls, [((), "done", 1)]) + + def test_async_run_requires_notify_model_and_method_together(self): + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.assertRaises(ValueError): + connection.with_context( + ai_connection_async=True, ai_connection_notify_model="res.partner" + )._run("Hello, AI!") + with self.assertRaises(ValueError): + connection.with_context( + ai_connection_async=True, + ai_connection_notify_method="_ai_connection_run_test_callback", + )._run("Hello, AI!") + + def test_sync_run_is_unaffected(self): + """Without ai_connection_async, behaviour must be exactly + `ai_connection`'s own - no `ai.connection.run` created, no job + enqueued. `_run_ai_step` isn't even overridden as far as this is + concerned; it just falls through to `super()` every round.""" + connection = self.env["ai.connection"].create( + {"name": "Demo Connection", "kind": "demo"} + ) + with self.trap_jobs() as trap: + response = connection._run("Hello, AI!") + trap.assert_jobs_count(0) + self.assertEqual( + response[0], "This is a demo response to the prompt: Hello, AI!" + ) + self.assertFalse(self.env["ai.connection.run"].search([]))