joungmin/vpd-permission-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refs #732: scope report titles to selected user

Browse Source
This commit is contained in:
devmrko
2026-08-10 18:02:49 +09:00
parent 84029ed633
commit cbd49b6b05
6 changed files with 225 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
ai-web-agent-console/app.py
View File

@@ -443,17 +443,28 @@ def _is_failed_synthesis_answer(value: object) -> bool:
)
def load_chat_context(conversation_id: str) -> list[dict[str, str]]:
def load_chat_context(
conversation_id: str,
*,
selected_user_id: str = "",
) -> list[dict[str, str]]:
current_user_id = str(selected_user_id or "").strip()
with _chat_db_connect() as connection:
rows = connection.execute(
"""
SELECT question, answer
FROM poc4_mcp_chat_turns
WHERE conversation_id = ?
AND (? = '' OR selected_user_id = ?)
ORDER BY turn_id DESC
LIMIT ?
""",
(conversation_id, CHAT_CONTEXT_TURNS),
(
conversation_id,
current_user_id,
current_user_id,
CHAT_CONTEXT_TURNS,
),
).fetchall()
messages: list[dict[str, str]] = []
for row in reversed(rows):
@@ -2997,6 +3008,7 @@ def _plan_presentation_title(
question: str,
source_step: Mapping[str, Any],
model_profile_key: str,
selected_user_id: str = "",
) -> str:
"""Generate a concise business title without coupling to a specific report."""
@@ -3013,12 +3025,17 @@ def _plan_presentation_title(
"Create one concise Korean enterprise report heading. Preserve the "
"business subject, team/user scope, and identifiers. Remove output-format "
"and action phrases such as HTML, report, dashboard, 보여줘, 만들어줘. "
"The current selected user ID is authoritative. First-person expressions "
"such as 내, 나, and 우리 refer to that current user, never to a user from "
"earlier conversation turns. Do not change the current user into a manager "
"or another employee unless the latest question explicitly names that scope. "
"Use a noun phrase, not a sentence. The template separately appends a fixed "
"subtitle, so do not include '최신 선사 실적'. Return only schema JSON."
),
user_prompt=json.dumps(
{
"question": question,
"current_selected_user_id": str(selected_user_id or "").strip(),
"source_tool": str(source_step.get("tool_name") or ""),
"source_rows": len(
_mcp_structured_rows(source_step.get("mcp_result", {}))
@@ -3048,6 +3065,7 @@ def _presentation_payload(
steps: list[Mapping[str, Any]],
*,
title: str = "",
selected_user_id: str = "",
) -> dict[str, Any]:
"""Build a renderer payload only from the preceding authorized tool result."""
@@ -3067,6 +3085,9 @@ def _presentation_payload(
if isinstance(row, Mapping)
]
requester = re.search(r"\bE\d{4,}\b", str(question or ""), flags=re.IGNORECASE)
requested_by = str(selected_user_id or "").strip().upper()
if not requested_by and requester:
requested_by = requester.group(0).upper()
source_tool = str(source.get("tool_name") or "MCP data query")
return {
"report": {
@@ -3074,7 +3095,7 @@ def _presentation_payload(
"category": "MCP Business Intelligence",
"title": _clean_presentation_title(title),
"question": str(question or ""),
"requestedBy": requester.group(0).upper() if requester else "-",
"requestedBy": requested_by or "-",
"generatedAt": datetime.now(timezone.utc).isoformat(),
"answer": (
f"{source_tool}에서 권한 범위 내 결과 {len(rows)}건을 받아 "
@@ -3100,6 +3121,7 @@ def _build_mcp_arguments(
preferred_tool: str,
steps: list[Mapping[str, Any]],
presentation_title: str = "",
selected_user_id: str = "",
) -> dict[str, Any]:
if not _is_presentation_route(route):
return build_mcp_tool_arguments(
@@ -3131,6 +3153,7 @@ def _build_mcp_arguments(
tool_query,
steps,
title=presentation_title,
selected_user_id=selected_user_id,
),
ensure_ascii=False,
)
@@ -3892,6 +3915,7 @@ def run_mcp_agent_loop(
bearer_token: str,
limit: int,
model_profile_key: str,
selected_user_id: str = "",
initial_route_key: str = "",
progress_callback: Any = None,
) -> dict[str, Any]:
@@ -4112,6 +4136,7 @@ def run_mcp_agent_loop(
question=question,
source_step=source_step,
model_profile_key=model_profile_key,
selected_user_id=selected_user_id,
)
arguments = _build_mcp_arguments(
route=route,
@@ -4120,6 +4145,7 @@ def run_mcp_agent_loop(
preferred_tool=server.default_tool,
steps=steps,
presentation_title=presentation_title,
selected_user_id=selected_user_id,
)
raw_result = call_tool(
base_url=server.endpoint_url,
@@ -4324,9 +4350,11 @@ def resolve_standalone_question(
question: str,
messages: list[Mapping[str, Any]],
model_profile_key: str,
selected_user_id: str = "",
) -> str:
context = _conversation_context(messages)
if not context:
current_user_id = str(selected_user_id or "").strip()
if not context and not current_user_id:
return question
try:
profile = resolve_model_profile(model_profile_key)
@@ -4339,10 +4367,19 @@ def resolve_standalone_question(
system_prompt=(
"Rewrite the user's latest Korean question into one standalone "
"MCP tool query. Use the conversation only to resolve references. "
"The current selected user ID is authoritative. First-person expressions "
"such as 내, 나, and 우리 refer only to the current selected user. Never "
"reuse another user's identity, manager scope, or team scope from earlier "
"conversation turns. If the latest question is self-contained, preserve its "
"meaning and scope. "
"Do not answer the question."
),
user_prompt=json.dumps(
{"conversation": context, "latest_question": question},
{
"current_selected_user_id": current_user_id,
"conversation": context,
"latest_question": question,
},
ensure_ascii=False,
),
response_schema={
@@ -5454,7 +5491,15 @@ def _process_submitted_question(
with processing_status:
refresh_progress(5, "대화 문맥을 불러오고 있습니다.")
step_started = perf_counter()
conversation = load_chat_context(conversation_id)
current_selected_user_id = (
selected_token_preset.user_id
if selected_token_preset is not None
else ""
)
conversation = load_chat_context(
conversation_id,
selected_user_id=current_selected_user_id,
)
step_elapsed = perf_counter() - step_started
refresh_progress(
15,
@@ -5469,6 +5514,7 @@ def _process_submitted_question(
question=normalized_question,
messages=conversation,
model_profile_key=active_model_profile,
selected_user_id=current_selected_user_id,
)
step_elapsed = perf_counter() - step_started
refresh_progress(
@@ -5591,6 +5637,7 @@ def _process_submitted_question(
bearer_token=bearer_token,
limit=int(limit),
model_profile_key=reasoning_model_profile,
selected_user_id=current_selected_user_id,
initial_route_key=_route_key(
selected_route.server_id,
selected_route.tool.name,

127
ai-web-agent-console/tests/test_scenarios.py
View File

@@ -6,6 +6,7 @@ import json
from pathlib import Path
import re
import tempfile
from types import SimpleNamespace
from typing import Any, Mapping
import unittest
from unittest.mock import patch
@@ -237,6 +238,132 @@ class DemoScenarioConfigTest(unittest.TestCase):
source,
)
def test_chat_context_is_scoped_to_current_selected_user(self) -> None:
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
tree = ast.parse(source)
helper = next(
node
for node in tree.body
if isinstance(node, ast.FunctionDef)
and node.name == "load_chat_context"
)
captured: dict[str, Any] = {}
class FakeConnection:
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def execute(self, sql, params):
captured["sql"] = sql
captured["params"] = params
return self
def fetchall(self):
return [{"question": "내 담당 선사", "answer": "2건"}]
namespace: dict[str, Any] = {
"CHAT_CONTEXT_TURNS": 8,
"MAX_CONVERSATION_MESSAGES": 16,
"_chat_db_connect": FakeConnection,
"_is_failed_synthesis_answer": lambda _value: False,
}
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
messages = namespace["load_chat_context"](
"conversation-1",
selected_user_id="E1002",
)
self.assertIn("selected_user_id = ?", captured["sql"])
self.assertEqual(
captured["params"],
("conversation-1", "E1002", "E1002", 8),
)
self.assertEqual(messages[0]["content"], "내 담당 선사")
def test_standalone_question_uses_current_user_without_prior_context(self) -> None:
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
tree = ast.parse(source)
helpers = [
node
for node in tree.body
if isinstance(node, ast.FunctionDef)
and node.name in {"_conversation_context", "resolve_standalone_question"}
]
captured: dict[str, Any] = {}
class FakeClient:
def complete(self, **kwargs):
captured.update(kwargs)
return json.dumps(
{
"standalone_question": (
"E1002 사용자의 담당 선사 최신 실적을 조회해줘"
)
},
ensure_ascii=False,
)
namespace: dict[str, Any] = {
"Any": Any,
"Mapping": Mapping,
"MAX_CONVERSATION_MESSAGES": 16,
"json": json,
"resolve_model_profile": lambda _key: SimpleNamespace(
model_id="model",
answer_model_region="region",
answer_model_endpoint="endpoint",
),
"build_oci_genai_completion_client": lambda *_args: FakeClient(),
"temperature_for_model_profile": lambda _profile: 0.0,
}
exec(compile(ast.Module(body=helpers, type_ignores=[]), "app.py", "exec"), namespace)
rewritten = namespace["resolve_standalone_question"](
question="내 담당 선사 최신 실적을 리포트로 보여줘",
messages=[],
model_profile_key="test",
selected_user_id="E1002",
)
prompt_payload = json.loads(captured["user_prompt"])
self.assertTrue(rewritten.startswith("E1002"))
self.assertEqual(prompt_payload["current_selected_user_id"], "E1002")
self.assertIn("authoritative", captured["system_prompt"])
def test_report_payload_requester_prefers_current_selected_user(self) -> None:
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
tree = ast.parse(source)
helper = next(
node
for node in tree.body
if isinstance(node, ast.FunctionDef)
and node.name == "_presentation_payload"
)
namespace: dict[str, Any] = {
"Any": Any,
"Mapping": Mapping,
"datetime": __import__("datetime").datetime,
"timezone": __import__("datetime").timezone,
"re": re,
"_mcp_structured_rows": lambda _value: [],
"_normalize_presentation_value": lambda value: value,
"_clean_presentation_title": lambda value: str(value),
}
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
payload = namespace["_presentation_payload"](
"E1001 팀장 문맥이 남은 질문",
[],
title="담당 선사 실적",
selected_user_id="E1002",
)
self.assertEqual(payload["report"]["requestedBy"], "E1002")
def test_hmm_report_data_query_drops_html_format_request(self) -> None:
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
tree = ast.parse(source)