mini-openclaw-v2/backend/api/workbench_configuration.py

"""Scoped configuration and read-only trajectory quality API (BR72–78)."""
from typing import Literal
from fastapi import APIRouter,Depends,Request,HTTPException
from pydantic import BaseModel,ConfigDict,Field
from api.risk_agent_access import authorize
from api.project_assistant import enabled
from api.lease_library import service,guard,project,card,translate as library_translate
from fastapi.responses import JSONResponse
from services import assistant_store as store,catalog_configuration as catalog,trajectory_quality as quality
router=APIRouter(prefix='/api/xingbang/risk-agent/project-assistant/library',dependencies=[Depends(authorize),Depends(enabled)])
def translate(call):
    def checked():
        try:return call()
        except ValueError as exc:
            text=str(exc)
            if 'COMPONENT_HIDDEN' in text:return JSONResponse(status_code=409,content={'error':{'code':'COMPONENT_HIDDEN','message':'评审方案存在隐藏组件，请更新配置后重试'}})
            return JSONResponse(status_code=422,content={'error':{'code':'CONFIGURATION_INVALID','message':'配置不合法，请检查必填项与受信模型地址'}})
    return library_translate(checked)

class Input(BaseModel):model_config=ConfigDict(extra='forbid')
class Visibility(Input):hidden:bool
class CardInput(Input):
    project_id:str=Field(min_length=1,max_length=180)
    kind:Literal['data_card','template']
    name:str=Field(min_length=1,max_length=120)
    content:dict
class Query(Input):
    project_id:str=Field(min_length=1,max_length=180)
    message:str=Field(min_length=1,max_length=1000)
class Settings(Input):
    provider:Literal['disabled','local','jev']
    endpoint:str=Field(default='',max_length=500)
    model:str=Field(default='',max_length=160)
    api_key:str|None=Field(default=None,max_length=4000)
    allow_external:bool=False
class RunInput(Input):run_id:str=Field(min_length=1,max_length=180)
@router.patch('/cards/{card_id}/visibility')
def visibility(card_id:str,payload:Visibility,request:Request):
    actor=guard(request,True)
    def action():
        with service().connect() as db:return catalog.set_visibility(db,card(db,request,card_id),actor,payload.hidden)
    return translate(action)
@router.post('/configuration/cards',status_code=201)
def create_card(payload:CardInput,request:Request):
    actor=guard(request,True)
    def action():
        with service().connect() as db:return catalog.create(db,project(db,request,payload.project_id),actor,payload.kind,payload.name,payload.content)
    return translate(action)
@router.post('/inventory')
def inventory(payload:Query,request:Request):
    guard(request)
    def action():
        with service().connect() as db:
            context=project(db,request,payload.project_id)
            return catalog.answer(db,context['id'],context['data_domain'],request.state.identity['role'],payload.message)
    return translate(action)
@router.get('/quality/settings')
def settings(request:Request):
    guard(request,True);return quality.public()
@router.put('/quality/settings')
def save_settings(payload:Settings,request:Request):
    actor=guard(request,True)
    def action():
        result=quality.save(**payload.model_dump())
        with service().connect() as db:store.event(db,'','quality.settings_updated',actor,{'provider':result['provider'],'revision':result['revision']})
        return result
    return translate(action)
@router.post('/quality/probe')
def probe(request:Request):
    guard(request,True);return quality.probe()
@router.get('/projects/{project_id}/quality')
def history(project_id:str,request:Request):
    guard(request)
    def action():
        with service().connect() as db:
            project(db,request,project_id)
            return {'items':store.listing(db,'quality_review',project_id)}
    return translate(action)
@router.post('/projects/{project_id}/quality',status_code=201)
def inspect(project_id:str,payload:RunInput,request:Request):
    actor=guard(request)
    def action():
        svc=service()
        with svc.connect() as db:
            project(db,request,project_id);run=store.get(db,'run',payload.run_id)
            if run['project_id']!=project_id:raise KeyError('运行不可访问')
        result=quality.inspect(run)
        with svc.connect() as db:
            store.put(db,'quality_review',result,immutable=True)
            store.event(db,project_id,'trajectory.quality',actor,{'review_id':result['id'],'run_id':payload.run_id,'status':result['judgment']['status']})
        return result
    return translate(action)

@router.get('/configuration/sources')
def sources(project_id:str,request:Request):
    guard(request,True)
    def action():
        from services import component_registry as registry
        with service().connect() as db:
            context=project(db,request,project_id)
            rows=[registry.resolve(db,p['id']) for p in store.listing(db,'release')]
            return {'items':[{'id':r['id'],'name':r['name'],'type':r['type']} for r in rows if r.get('kind')=='data_source' and r.get('scope')==context['data_domain'] and r.get('project_id') in (None,'',project_id) and not catalog.hidden(db,r['id'])]}
    return translate(action)

@router.get('/projects/{project_id}/runs/{run_id}/influences')
def influences(project_id:str,run_id:str,request:Request):
    guard(request)
    def action():
        from services.prescreen_explanation import explain
        with service().connect() as db:
            project(db,request,project_id);run=store.get(db,'run',run_id)
            if run.get('project_id')!=project_id:raise KeyError('运行不可访问')
            view=store.get(db,'prescreen_run',run_id)
            return explain(run,view)
    return translate(action)