Files
cowork-local/infrastructure/config/settings_facade.py
T
f9f6bc01fd
CI / test (push) Canceled after 0s
Feature/delta team/epic r04 (#7)
## Summary

epic r04 - begin refactor

## Change Type

- [x] Cowork feature
- [ ] Bug fix
- [ ] Core AI contribution
- [ ] Test / hardening
- [ ] Performance
- [ ] Documentation

## Related Work

Cowork Task:

Core Repo: http://34.143.229.138/gitea-admin/fsg-ai-core-assets

Core AI Issue:

Core Task:

Related PR:

## Scope

What is intentionally included?

What is intentionally NOT included?

## Validation

- [ ] Unit tests
- [ ] Integration tests
- [ ] Manual verification
- [ ] Regression check

Commands / evidence:

## Security Impact

Permission / credential / network / customer data impact:

## Compatibility

- [ ] No breaking change
- [ ] Breaking change documented

## Reviewer Notes

Anything Cowork reviewers should pay attention to.

---------

Co-authored-by: Anh Tran Nguyen Minh <anhtnm1@fpt.com>
Co-authored-by: Huong Le Thi Thien <huongltt35@fpt.com>
Co-authored-by: Nam Pham Dinh Thanh <nampdt@fpt.com>
Co-authored-by: Vu Dam Tuan <vudt15@fpt.com>
Co-authored-by: Hiep Ha Van <hiephv3@fpt.com>
Co-authored-by: Lam Hoang Van <lamhv7@fpt.com>
Reviewed-on: #7
Co-authored-by: Duy Le Huu <duylh19@fpt.com>
2026-08-31 05:15:13 +00:00

217 lines
8.5 KiB
Python

"""Khung nhìn có kiểu cho từng nhóm cấu hình — R02-T03.
Vấn đề đang có: khắp nơi viết ``ctx.config.routing.get("switch_mode", "off")``.
Gõ sai một chữ thì lặng lẽ nhận giá trị mặc định, không ai biết cho tới khi
tính năng "không hiểu sao không chạy". Đếm được **156 lời gọi ``ctx.config.*``
trong 29 file** kiểu đó.
Ở đây mỗi nhóm cấu hình có một lớp: gõ sai tên thuộc tính là lỗi ngay, và kiểu
dữ liệu ghi rõ ràng nên đọc code là biết ``confirm_timeout_sec`` là số giây
chứ không phải mili giây.
Cố ý KHÔNG dùng dataclass đông cứng: đây là *khung nhìn* lên dict cấu hình
sống, sửa qua đây là sửa vào dict rồi ``save()`` là xuống đĩa. Sao chép thành
dataclass thì lại sinh chuyện đồng bộ hai chiều.
SEAM · dựng 2026-08-22 · chưa nối dây (F-05)
------------------------------------------------------------
Được nối khi: ít nhất một trong 156 lời gọi ``ctx.config.*`` chuyển sang đọc qua ``Settings``.
Để dormant thì sao: Mục đích của nó là chặn lỗi gõ sai tên khoá. Không ai
dùng thì không chặn được gì cả.
Cổng ``scripts/check_orphan_modules.py`` đếm tuổi seam từ ngày trên
và nhắc khi quá ``SEAM_MAX_AGE_DAYS``. Đổi nội dung dòng đó thì cổng
đọc theo — đừng sửa ngày để làm im lời nhắc.
"""
from __future__ import annotations
from typing import Any, Dict
class _View:
"""Khung nhìn lên một nhánh của dict cấu hình."""
def __init__(self, data: Dict[str, Any]):
"""Bọc một nhánh dict cấu hình để đọc bằng thuộc tính thay vì tra khoá."""
self._d = data
def _get(self, key: str, default: Any) -> Any:
"""Đọc một khoá, coi ``None`` như thiếu.
Cấu hình cũ có chỗ ghi ``null``; nếu trả thẳng ``None`` ra ngoài thì
``str(None)`` thành chuỗi "None" và lỗi hiện ra ở tận nơi dùng.
"""
value = self._d.get(key, default)
return default if value is None else value
def raw(self) -> Dict[str, Any]:
"""Dict gốc — dùng khi cần đọc khoá chưa được đưa vào khung nhìn.
Có mặt để không ai bị kẹt: thiếu thuộc tính thì dùng tạm ``raw()`` rồi
mở issue bổ sung, chứ đừng vòng lại ``ctx.config.data``.
"""
return self._d
class ProviderSettings(_View):
"""Một provider: đi đâu, model nào, khoá nào.
``api_key`` ở đây là thứ ``JsonConfigRepository.provider_conf()`` đã ghép
sẵn từ kho bí mật — xem đường A trong ``GammaTeam_decisions.md``.
"""
@property
def base_url(self) -> str:
"""Endpoint của provider. '' nghĩa là chưa cấu hình."""
return str(self._get("base_url", ""))
@property
def model(self) -> str:
"""Model mặc định của provider này. '' nghĩa là chưa chọn."""
return str(self._get("model", ""))
@property
def api_key(self) -> str:
"""Khoá API — đã được ``provider_conf()`` ghép từ kho bí mật của hệ điều hành."""
return str(self._get("api_key", ""))
@property
def configured(self) -> bool:
"""Đủ thông tin để gọi được chưa.
Ollama chạy cục bộ nên không cần khoá — đó là lý do điều kiện là
"có base_url và model", không phải "có api_key".
"""
return bool(self.base_url and self.model)
class RoutingSettings(_View):
"""Định tuyến model tự động (``core/routing/``)."""
@property
def switch_mode(self) -> str:
"""``"off"`` | ``"auto"`` | ``"manual"``."""
return str(self._get("switch_mode", "off"))
@switch_mode.setter
def switch_mode(self, value: str) -> None:
"""Đặt chế độ định tuyến chung. Sửa thẳng vào dict cấu hình sống."""
self._d["switch_mode"] = value
@property
def enabled(self) -> bool:
"""Định tuyến tự động có đang bật không (tức ``switch_mode`` khác "off")."""
return self.switch_mode != "off"
@property
def policy(self) -> str:
"""``"balanced"`` | ``"cheap"`` | ``"quality"``…"""
return str(self._get("policy", "balanced"))
@property
def min_score_gain(self) -> float:
"""Phải hơn model hiện tại bao nhiêu điểm mới đáng đổi."""
return float(self._get("min_score_gain", 0.05))
@property
def confirm_timeout_sec(self) -> int:
"""GIÂY, không phải mili giây — đọc tên là biết, khỏi phải mò."""
return int(self._get("confirm_timeout_sec", 60))
@property
def reassess_interval_hours(self) -> int:
"""GIỜ giữa hai lần chấm điểm lại danh mục model."""
return int(self._get("reassess_interval_hours", 24))
@property
def per_provider_concurrency(self) -> int:
"""Số lượt gọi chạy song song tối đa cho mỗi provider khi dò/chấm điểm."""
return int(self._get("per_provider_concurrency", 2))
@property
def judge_provider(self) -> str:
"""Provider dùng làm trọng tài chấm điểm model. '' nghĩa là dùng provider đang chọn."""
return str(self._get("judge_provider", ""))
@property
def judge_model(self) -> str:
"""Model dùng làm trọng tài chấm điểm. '' nghĩa là dùng model mặc định của trọng tài."""
return str(self._get("judge_model", ""))
class SecuritySettings(_View):
"""Chính sách an toàn cho agent (``core/agent_security.py``)."""
@property
def enabled(self) -> bool:
"""Công tắc tổng của lớp an toàn. Tắt là bỏ qua mọi bước kiểm dưới đây."""
return bool(self._get("enabled", True))
@property
def validate_prompt(self) -> bool:
"""Có quét prompt người dùng tìm dấu hiệu tấn công tiêm lệnh không."""
return bool(self._get("validate_prompt", True))
@property
def validate_attachments(self) -> bool:
"""Có kiểm tệp đính kèm (đuôi/kiểu MIME nguy hiểm) trước khi đưa vào lượt chat không."""
return bool(self._get("validate_attachments", True))
@property
def validate_commands(self) -> bool:
"""Có phân loại rủi ro lệnh shell trước khi chạy không."""
return bool(self._get("validate_commands", True))
@property
def command_ai_check(self) -> bool:
"""Có nhờ thêm AI xét lệnh khi bộ luật tĩnh chưa chắc chắn không. Mặc định tắt vì tốn một lượt gọi."""
return bool(self._get("command_ai_check", False))
@property
def cowork_confirm_commands(self) -> bool:
"""Có hỏi trước khi chạy lệnh không.
Ứng với ``PolicyOutcome.ASK`` trong
``domain/security/tool_policy.py``.
"""
return bool(self._get("cowork_confirm_commands", True))
@property
def rules_onedrive_url(self) -> str:
"""Link OneDrive tới bộ luật an toàn dùng chung. '' nghĩa là dùng bản đóng gói sẵn trong app."""
return str(self._get("rules_onedrive_url", ""))
@property
def admin_email(self) -> str:
"""Email quản trị nhận cảnh báo vi phạm. '' nghĩa là không gửi."""
return str(self._get("admin_email", ""))
class Settings:
"""Cửa vào duy nhất cho các nhóm cấu hình có kiểu.
>>> s = Settings(repo)
>>> if s.routing.enabled and s.provider().configured:
... ...
"""
def __init__(self, repo):
"""Bọc một ``ConfigRepository`` — mọi lượt đọc/ghi đều đi xuống nó, lớp này chỉ
đổi cách gọi cho dễ đọc.
"""
self._repo = repo
def provider(self, name: str | None = None) -> ProviderSettings:
"""Khung nhìn cấu hình của một provider; bỏ trống thì lấy provider đang chọn."""
return ProviderSettings(self._repo.provider_conf(name))
@property
def routing(self) -> RoutingSettings:
"""Khung nhìn nhóm cấu hình định tuyến model."""
return RoutingSettings(self._repo.routing)
@property
def security(self) -> SecuritySettings:
"""Khung nhìn nhóm cấu hình an toàn cho agent."""
return SecuritySettings(self._repo.agent_security)