Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion backend/app/api/endpoints/adapter/tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,11 @@
from app.api.dependencies import get_db, with_task_telemetry
from app.core import security
from app.core.config import settings
from app.core.constants import CLIENT_ORIGIN_FRONTEND, SUPPORTED_CLIENT_ORIGINS
from app.core.constants import (
CLIENT_ORIGIN_FRONTEND,
CLIENT_ORIGIN_WEWORK,
SUPPORTED_CLIENT_ORIGINS,
)
from app.db.session import get_async_db
from app.models.user import User
from app.schemas.remote_workspace import (
Expand Down Expand Up @@ -57,6 +61,9 @@
from app.services import prompt_draft_service
from app.services.adapters.executor_job import job_service
from app.services.adapters.task_kinds import task_kinds_service
from app.services.adapters.wework_conversation_search import (
search_wework_conversation_tasks,
)
from app.services.chat.storage import session_manager
from app.services.remote_workspace_service import remote_workspace_service
from app.services.shared_task import shared_task_service
Expand Down Expand Up @@ -274,6 +281,29 @@ def search_tasks_by_title(
return {"total": total, "items": items}


@router.get("/wework/conversation-search", response_model=TaskListResponse)
def search_wework_conversation_task_list(
keyword: str = Query(
..., min_length=1, description="Search by task title or message keywords"
),
page: int = Query(1, ge=1, description="Page number"),
limit: int = Query(10, ge=1, le=100, description="Items per page"),
current_user: User = Depends(security.get_current_user),
db: Session = Depends(get_db),
):
"""Search WeWork conversations by title or conversation content."""
skip = (page - 1) * limit
items, total = search_wework_conversation_tasks(
db=db,
user_id=current_user.id,
keyword=keyword,
skip=skip,
limit=limit,
client_origin=CLIENT_ORIGIN_WEWORK,
)
return {"total": total, "items": items}


@router.get("/archived", response_model=ArchivedTaskListResponse)
def get_archived_tasks(
page: int = Query(1, ge=1, description="Page number"),
Expand Down
216 changes: 216 additions & 0 deletions backend/app/services/adapters/wework_conversation_search.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,216 @@
# SPDX-FileCopyrightText: 2026 Weibo, Inc.
#
# SPDX-License-Identifier: Apache-2.0

"""WeWork-specific conversation search service."""

import json
from typing import Any, Dict, List, Tuple

from sqlalchemy import String, cast, or_
from sqlalchemy.orm import Session

from app.core.constants import CLIENT_ORIGIN_WEWORK
from app.models.subtask import Subtask
from app.models.task import TaskResource
from app.schemas.kind import Task
from app.services.adapters.task_kinds.converters import convert_to_task_dict_optimized
from app.services.adapters.task_kinds.filters import filter_tasks_for_display
from app.services.adapters.task_kinds.helpers import get_tasks_related_data_batch
from app.services.adapters.task_kinds.query_utils import restore_task_order

SEARCH_RESULT_FALLBACK_TASK_LIMIT = 300
SEARCH_RESULT_FALLBACK_SUBTASK_LIMIT = 1000


def _stringify_search_value(value: Any) -> str:
if value is None:
return ""
if isinstance(value, str):
return value
return json.dumps(value, ensure_ascii=False, default=str)


def _get_subtask_search_text(subtask: Subtask) -> str:
return "\n".join(
[
_stringify_search_value(subtask.prompt),
_stringify_search_value(subtask.result),
_stringify_search_value(subtask.error_message),
]
)


def search_wework_conversation_tasks(
*,
db: Session,
user_id: int,
keyword: str,
skip: int = 0,
limit: int = 100,
client_origin: str = CLIENT_ORIGIN_WEWORK,
) -> Tuple[List[Dict[str, Any]], int]:
"""Search WeWork-owned tasks by title or conversation content."""
Comment on lines +44 to +53

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win

Add an OpenTelemetry tracing decorator to this entrypoint.

search_wework_conversation_tasks is a public service entrypoint but is not traced. The new endpoint search_wework_conversation_task_list is likewise untraced. As per coding guidelines: "Use OpenTelemetry tracing decorators: @trace_async or @trace_sync for entire functions".

Proposed change
+from shared.telemetry.decorators import trace_sync
+
+@trace_sync("search_wework_conversation_tasks", "tasks.search")
 def search_wework_conversation_tasks(
     *,
     db: Session,
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/app/services/adapters/wework_conversation_search.py` around lines 44
- 53, This function is a public service entrypoint that lacks OpenTelemetry
tracing; add the tracing decorator `@trace_sync` to both
search_wework_conversation_tasks and the new
search_wework_conversation_task_list (they are synchronous), and also add the
corresponding import for trace_sync at the top of the module so the decorator is
available. Ensure the decorator wraps the entire function definition (above def)
so the entrypoints are traced end-to-end.

keyword = keyword.strip()
if not keyword:
return [], 0

task_ids, visible_task_by_id = _load_visible_wework_tasks(
db=db,
user_id=user_id,
client_origin=client_origin,
)
if not task_ids:
return [], 0

matched_task_ids, ordered_matched_task_ids = _collect_matched_task_ids(
db=db,
task_ids=task_ids,
visible_task_by_id=visible_task_by_id,
keyword=keyword,
required_count=skip + limit,
)
id_to_task = {
task_id: visible_task_by_id[task_id]
for task_id in matched_task_ids
if task_id in visible_task_by_id
}
total = len(id_to_task)
filtered_tasks = restore_task_order(
ordered_matched_task_ids[skip : skip + limit], id_to_task, limit
)
if not filtered_tasks:
return [], total

return _format_task_results(db, filtered_tasks, user_id), total


def _collect_matched_task_ids(
*,
db: Session,
task_ids: List[int],
visible_task_by_id: Dict[int, TaskResource],
keyword: str,
required_count: int,
) -> Tuple[set[int], List[int]]:
keyword_lower = keyword.lower()
matched_task_ids = _match_task_titles(visible_task_by_id, keyword_lower)
matched_task_ids.update(_match_subtasks_in_database(db, task_ids, keyword))
ordered_matched_task_ids = [
task_id for task_id in task_ids if task_id in matched_task_ids
]

if len(ordered_matched_task_ids) < required_count:
matched_task_ids.update(
_match_subtasks_with_python_fallback(
db=db,
task_ids=task_ids,
matched_task_ids=matched_task_ids,
keyword_lower=keyword_lower,
)
)
ordered_matched_task_ids = [
task_id for task_id in task_ids if task_id in matched_task_ids
]

return matched_task_ids, ordered_matched_task_ids
Comment on lines +66 to +116

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
ast-grep --pattern 'def restore_task_order($$$):
  $$$'
rg -nP --type=py -C3 '\brestore_task_order\b'

Repository: wecode-ai/Wegent

Length of output: 6303


🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Show relevant part of wework_conversation_search.py (around the snippet)
sed -n '1,140p' backend/app/services/adapters/wework_conversation_search.py

# Show the fallback/matching helpers referenced in the snippet
rg -n "def _collect_matched_task_ids|_match_subtasks_in_database|_match_subtasks_with_python_fallback|def _match_task_titles" backend/app/services/adapters/wework_conversation_search.py
sed -n '140,320p' backend/app/services/adapters/wework_conversation_search.py

# Inspect restore_task_order implementation
sed -n '360,460p' backend/app/services/adapters/task_kinds/query_utils.py

Repository: wecode-ai/Wegent

Length of output: 8141


Stabilize pagination by making the matched-task set (and total) independent of skip

  • _collect_matched_task_ids() runs _match_subtasks_with_python_fallback() only when len(ordered_matched_task_ids) < (skip + limit), so matched_task_ids (and therefore total = len(id_to_task)) changes between page 1 and page 2.
  • Since search_wework_conversation_tasks() slices ordered_matched_task_ids[skip : skip + limit] using that page-dependent matched set, tasks can be duplicated or skipped across pages when fallback gets added on later page requests.
  • restore_task_order() does not reorder—it only projects the provided task_ids slice in order until limit, so the instability originates from the fallback gating.

Compute fallback/matched IDs once using a skip-independent criterion (e.g., always apply fallback for the same bounded ID set like task_ids[:SEARCH_RESULT_FALLBACK_TASK_LIMIT], or gate on a fixed cap not based on page offset) and derive both ordered_matched_task_ids and total from that same final matched set.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/app/services/adapters/wework_conversation_search.py` around lines 66
- 116, The current bug is that _collect_matched_task_ids() gates calling
_match_subtasks_with_python_fallback() on a page-dependent required_count
(skip+limit), causing matched_task_ids/total to differ between pages; fix by
making the fallback deterministic and skip-independent: introduce a fixed
SEARCH_RESULT_FALLBACK_TASK_LIMIT constant (or hardcap value) and always run
_match_subtasks_with_python_fallback() against
task_ids[:SEARCH_RESULT_FALLBACK_TASK_LIMIT] (or pass that cap into the
function) to extend matched_task_ids, then compute ordered_matched_task_ids from
the full task_ids using that final matched_task_ids so total and paging are
stable; update the _collect_matched_task_ids() logic to remove the
skip-dependent gating and reference _match_subtasks_with_python_fallback,
_match_subtasks_in_database, _match_task_titles, and restore_task_order
accordingly.



def _load_visible_wework_tasks(
*,
db: Session,
user_id: int,
client_origin: str,
) -> Tuple[List[int], Dict[int, TaskResource]]:
tasks = (
db.query(TaskResource)
.filter(
TaskResource.kind == "Task",
TaskResource.is_active == TaskResource.STATE_ACTIVE,
TaskResource.namespace != "system",
TaskResource.user_id == user_id,
TaskResource.client_origin == client_origin,
)
.order_by(TaskResource.updated_at.desc())
.all()
)
visible_task_by_id = filter_tasks_for_display(tasks)
task_ids = [task.id for task in tasks if task.id in visible_task_by_id]
return task_ids, visible_task_by_id


def _format_task_results(
db: Session,
filtered_tasks: List[TaskResource],
user_id: int,
) -> List[Dict[str, Any]]:
related_data_batch = get_tasks_related_data_batch(db, filtered_tasks, user_id)
result = []
for task in filtered_tasks:
task_crd = Task.model_validate(task.json)
task_related_data = related_data_batch.get(str(task.id), {})
result.append(convert_to_task_dict_optimized(task, task_related_data, task_crd))

return result


def _match_task_titles(
visible_task_by_id: Dict[int, TaskResource],
keyword_lower: str,
) -> set[int]:
return {
task.id
for task in visible_task_by_id.values()
if keyword_lower in (Task.model_validate(task.json).spec.title or "").lower()
}


def _match_subtasks_in_database(
db: Session,
task_ids: List[int],
keyword: str,
) -> set[int]:
like_pattern = f"%{keyword}%"
rows = (
db.query(Subtask.task_id)
.filter(Subtask.task_id.in_(task_ids))
.filter(
or_(
Subtask.prompt.ilike(like_pattern),
Subtask.error_message.ilike(like_pattern),
cast(Subtask.result, String).ilike(like_pattern),
)
)
.distinct()
.all()
)
return {row[0] for row in rows}
Comment on lines +168 to +187

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Escape LIKE wildcards in the keyword.

% and _ in the user keyword are treated as wildcards inside ilike, so a query like 100% or a_b matches more broadly than the literal text the user typed. Escape them (and set an escape char) before interpolating into like_pattern.

Proposed change
-    like_pattern = f"%{keyword}%"
+    escaped = keyword.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+    like_pattern = f"%{escaped}%"
     rows = (
         db.query(Subtask.task_id)
         .filter(Subtask.task_id.in_(task_ids))
         .filter(
             or_(
-                Subtask.prompt.ilike(like_pattern),
-                Subtask.error_message.ilike(like_pattern),
-                cast(Subtask.result, String).ilike(like_pattern),
+                Subtask.prompt.ilike(like_pattern, escape="\\"),
+                Subtask.error_message.ilike(like_pattern, escape="\\"),
+                cast(Subtask.result, String).ilike(like_pattern, escape="\\"),
             )
         )
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/app/services/adapters/wework_conversation_search.py` around lines 168
- 187, In _match_subtasks_in_database: escape SQL LIKE wildcards in the incoming
keyword before building like_pattern by first replacing backslashes with double
backslashes then prefixing any '%' and '_' with a backslash (or chosen escape
char), build like_pattern = f"%{escaped_keyword}%", and pass the same escape
char into the ilike calls (e.g., .ilike(like_pattern, escape='\\')) for
Subtask.prompt, Subtask.error_message and cast(Subtask.result, String) so user
input is treated literally.



def _match_subtasks_with_python_fallback(
*,
db: Session,
task_ids: List[int],
matched_task_ids: set[int],
keyword_lower: str,
) -> set[int]:
fallback_task_ids = [
task_id
for task_id in task_ids[:SEARCH_RESULT_FALLBACK_TASK_LIMIT]
if task_id not in matched_task_ids
]
if not fallback_task_ids:
return set()

subtasks = (
db.query(Subtask)
.filter(Subtask.task_id.in_(fallback_task_ids))
.order_by(Subtask.updated_at.desc())
.limit(SEARCH_RESULT_FALLBACK_SUBTASK_LIMIT)
.all()
)
return {
subtask.task_id
for subtask in subtasks
if keyword_lower in _get_subtask_search_text(subtask).lower()
}
Loading
Loading