chenxin
c63b75152a
- 天工阁域 domains/tiangong：backstage_tasks（委托包+验收标准+结果证据）；有限分解（parent 链）与受控并行（各自后台 Run）；子任务全终态级联父任务进入待验收；按证据 accept/reject - kernel.start_backstage_run（kind=background_task）：loop 后台分支不占前台席位、与前台对话并行；断言放宽容纳无人物功能任务；结果只经回调回呈任务——无身份写入路径 - 叩心域 domains/kouxin：knocking_candidates（event/time_anchor/manual 触发 + 草稿 + 裁决快照）；四约束裁决（总开关→静默窗口【本地时区判定】→24h 频率→预算），dispatch 未过 429 且快照留痕；候选默认不自动发送；expire_stale 过期推进不触碰缘契 - 辰正 domains/zhengchen：temporal_context() 复用 time_sense 计算核（快照/农历/季节），产出【时态感知】投影段接入镜树生成 prompt（客观时间语境，不附情绪臆测——守一纪律） - 退役：modules/goal_orchestrator|planner|perception|big_cycle、application/planning、services/governance|goal_persist、candidates 表链（imports 候选抽取 API 面）、goals/plans/agent_roster/perception_channels/big_cycles/旧候选表（迁移 009）；前端 admin goals/plans/agents/perception/world 页与六组件；governance 只留 GenerationPreferences；admin 治理面收敛为司命+百工两枢；pivots 健康以域表计数重建 - 验收（test_m8_tiangong 五用例）：后台任务无身份写入（记忆/魂籍/缘契零新增）；后台与前台并行；分解级联；叩心四约束逐项；不回应不降低关系（缘契维度/状态/版本零变化） - 四套件：后端 345、前端 203、BFF 4、神识 5 包全绿（2 个预存环境失败不变）；迁移 009 净库干净
354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""司命台管理聚合服务（api-spec §19.1–§19.3）。
|
||
|
||
九枢健康度由各枢运行/治理数据派生（不落长期状态，§19.1）；朱批待办复用
|
||
待决请命（§19.2）；起居注审计复用 audit_events（§19.3）。
|
||
"""
|
||
|
||
import json
|
||
|
||
from sqlalchemy import func, select
|
||
|
||
from app.core.config import settings
|
||
from app.core.security import redact_text
|
||
from app.core.time import to_iso, utc_now
|
||
from app.db.session import AsyncSessionLocal
|
||
from app.models.idempotency import IdempotencyKey
|
||
from app.models.xirang import Memory
|
||
from app.models.runtime import Event, Run
|
||
from app.models.safety import AuditEvent, PendingConfirmationRecord
|
||
from app.repositories import audit as audit_repo
|
||
from app.repositories import characters as character_repo
|
||
from app.repositories import idempotency as idempotency_repo
|
||
from app.repositories import runtime as runtime_repo
|
||
from app.schemas.admin import (
|
||
AuditEntry,
|
||
PendingApproval,
|
||
PivotHealthReport,
|
||
PivotHealthSummary,
|
||
TraceDetail,
|
||
TraceItem,
|
||
TraceListResponse,
|
||
TraceSummary,
|
||
)
|
||
from app.schemas.common import Page
|
||
from app.services.serializers import audit_to_response
|
||
|
||
# 九枢全集（契约字典 §5.9）：枢标识、中文名、工程模块名。
|
||
_PIVOT_REGISTRY = [
|
||
("personality_core", "灵核枢", "Personality Core"),
|
||
("capability_registry", "百工枢", "Capability Registry"),
|
||
("safety_governor", "司命枢", "Safety Governor"),
|
||
("hetu", "河图枢", "Hetu World"),
|
||
("tiangong", "天工阁", "Tiangong Backstage"),
|
||
("kouxin", "叩心枢", "Kouxin Proactive"),
|
||
("memory", "息壤枢", "Xirang Memory"),
|
||
]
|
||
|
||
|
||
def _summarize_input(value: dict | None) -> str:
|
||
if not value:
|
||
return ""
|
||
text = json.dumps(value, ensure_ascii=False, default=str)
|
||
return redact_text(text)[:200]
|
||
|
||
|
||
async def get_pivots_health(
|
||
*,
|
||
owner_scope: str = settings.default_owner_scope,
|
||
):
|
||
"""九枢健康度：以事件流与域表计数呈现（旧枢治理表随 M8 退役）。"""
|
||
from app.domains.baigong.models import Capability
|
||
from app.domains.hetu.models import PersistentWorld
|
||
from app.domains.kouxin.models import KnockingCandidate
|
||
from app.domains.tiangong.models import BackstageTask
|
||
from app.models.safety import PendingConfirmationRecord
|
||
from app.models.xirang import Memory
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
pending_count = len(
|
||
list(
|
||
(
|
||
await session.execute(
|
||
select(PendingConfirmationRecord).where(
|
||
PendingConfirmationRecord.status == "pending"
|
||
)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
)
|
||
caps_n = await _count(session, Capability)
|
||
world_n = await _count(session, PersistentWorld)
|
||
tasks_n = await _count(session, BackstageTask)
|
||
knocks_n = await _count(session, KnockingCandidate)
|
||
memories_n = await _count(session, Memory, Memory.status != "tombstoned")
|
||
character = await character_repo.get_active_character(session, owner_scope)
|
||
has_personality = bool(character and character.identity_baseline_json not in ("", "{}"))
|
||
now = to_iso(utc_now())
|
||
|
||
raw: list[tuple[str, str, str]] = [
|
||
("personality_core", "steady" if has_personality else "void", "道体锚点已就绪" if has_personality else "锚点未初始化"),
|
||
("capability_registry", "steady" if caps_n else "void", f"能力 {caps_n} 项"),
|
||
("safety_governor", "cracked" if pending_count else "steady", f"{pending_count} 项待裁决" if pending_count else "无待裁决"),
|
||
("hetu", "steady" if world_n else "void", f"境界 {world_n} 个"),
|
||
("tiangong", "steady" if tasks_n else "void", f"受托任务 {tasks_n} 项"),
|
||
("kouxin", "breathing" if knocks_n else "void", f"叩心候选 {knocks_n} 条"),
|
||
("memory", "steady" if memories_n else "void", f"记忆 {memories_n} 条"),
|
||
]
|
||
|
||
name_map = {pivot: (name, module) for pivot, name, module in _PIVOT_REGISTRY}
|
||
pivots = [
|
||
PivotHealthSummary(
|
||
pivot=pivot, name=name_map[pivot][0], module=name_map[pivot][1],
|
||
health=health, summary=summary, updated_at=now,
|
||
)
|
||
for pivot, health, summary in raw
|
||
]
|
||
alerts = sum(1 for item in pivots if item.health == "cracked")
|
||
return PivotHealthReport(
|
||
overall="cracked" if alerts else "steady", alerts=alerts, pivots=pivots
|
||
)
|
||
|
||
|
||
async def _count(session, model, *criteria) -> int:
|
||
from sqlalchemy import func as _func
|
||
|
||
stmt = select(_func.count()).select_from(model)
|
||
for c in criteria:
|
||
stmt = stmt.where(c)
|
||
return int((await session.execute(stmt)).scalar_one())
|
||
|
||
async def get_pending_approvals() -> list[PendingApproval]:
|
||
async with AsyncSessionLocal() as session:
|
||
records = (
|
||
(
|
||
await session.execute(
|
||
select(PendingConfirmationRecord)
|
||
.where(PendingConfirmationRecord.status == "pending")
|
||
.order_by(PendingConfirmationRecord.created_at.asc())
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
approvals: list[PendingApproval] = []
|
||
for record in records:
|
||
try:
|
||
input_payload = json.loads(record.input_json) if record.input_json else {}
|
||
except ValueError:
|
||
input_payload = {}
|
||
approvals.append(
|
||
PendingApproval(
|
||
confirmation_id=record.id,
|
||
character_id=record.character_id,
|
||
tool=record.tool,
|
||
risk_level=record.risk_level,
|
||
permission_policy="ask",
|
||
action_summary=f"调用能力 {record.tool}",
|
||
input_summary=_summarize_input(input_payload),
|
||
possible_impact=None,
|
||
reason=record.reason,
|
||
created_at=record.created_at.isoformat() if record.created_at else "",
|
||
)
|
||
)
|
||
return approvals
|
||
|
||
|
||
async def get_audit_timeline(
|
||
*,
|
||
pivot: str | None = None,
|
||
page: int = 1,
|
||
page_size: int = 50,
|
||
) -> Page[AuditEntry]:
|
||
async with AsyncSessionLocal() as session:
|
||
events, total = await audit_repo.list_audit_events(
|
||
session, pivot=pivot, page=page, page_size=page_size
|
||
)
|
||
return Page(
|
||
items=[audit_to_response(event) for event in events],
|
||
page=page,
|
||
page_size=page_size,
|
||
total=total,
|
||
)
|
||
|
||
|
||
# —— §19.5 链路追踪（按 trace_id 聚合 events / audit_events / idempotency_keys）——
|
||
|
||
|
||
def _event_to_item(event: Event) -> TraceItem:
|
||
try:
|
||
payload = json.loads(event.payload_json) if event.payload_json else {}
|
||
except (json.JSONDecodeError, TypeError):
|
||
payload = {}
|
||
return TraceItem(
|
||
kind="event",
|
||
id=event.id,
|
||
created_at=event.created_at,
|
||
source_label=f"运行事件 / {event.type}",
|
||
summary=event.type,
|
||
detail={
|
||
"run_id": event.run_id,
|
||
"source": event.source,
|
||
"type": event.type,
|
||
"payload": payload,
|
||
"sequence": event.sequence,
|
||
"parent_id": event.parent_id,
|
||
},
|
||
)
|
||
|
||
|
||
def _audit_to_item(audit: AuditEvent) -> TraceItem:
|
||
parts = [audit.decision, audit.risk_level]
|
||
if audit.reason:
|
||
parts.append(audit.reason[:60])
|
||
return TraceItem(
|
||
kind="audit",
|
||
id=audit.id,
|
||
created_at=audit.created_at,
|
||
source_label=f"安全审计 / {audit.action}",
|
||
summary=" · ".join(parts),
|
||
detail={
|
||
"pivot": audit.pivot,
|
||
"action": audit.action,
|
||
"actor": audit.actor,
|
||
"decision": audit.decision,
|
||
"risk_level": audit.risk_level,
|
||
"reason": audit.reason,
|
||
"resource_type": audit.resource_type,
|
||
"resource_id": audit.resource_id,
|
||
},
|
||
)
|
||
|
||
|
||
def _idempotency_to_item(idem: IdempotencyKey) -> TraceItem:
|
||
return TraceItem(
|
||
kind="idempotency",
|
||
id=idem.id,
|
||
created_at=idem.created_at,
|
||
source_label=f"请求幂等 / {idem.method}",
|
||
summary=f"{idem.method} {idem.path} → {idem.response_status_code}",
|
||
detail={
|
||
"method": idem.method,
|
||
"path": idem.path,
|
||
"status": idem.status,
|
||
"response_status_code": idem.response_status_code,
|
||
"scope": idem.scope,
|
||
},
|
||
)
|
||
|
||
|
||
async def get_trace(trace_id: str) -> TraceDetail:
|
||
"""按 trace_id 聚合 events / audit_events / idempotency_keys 三表为升序时间线。"""
|
||
async with AsyncSessionLocal() as session:
|
||
events = await runtime_repo.list_events_by_trace(session, trace_id)
|
||
audits = await audit_repo.list_audit_events_by_trace(session, trace_id)
|
||
idempotencies = await idempotency_repo.list_by_trace(session, trace_id)
|
||
|
||
items: list[TraceItem] = [_event_to_item(event) for event in events]
|
||
items.extend(_audit_to_item(audit) for audit in audits)
|
||
items.extend(_idempotency_to_item(idem) for idem in idempotencies)
|
||
items.sort(key=lambda item: item.created_at)
|
||
|
||
run_ids = sorted({event.run_id for event in events if event.run_id})
|
||
character_ids = sorted(
|
||
{
|
||
*[event.character_id for event in events if event.character_id],
|
||
*[audit.character_id for audit in audits if audit.character_id],
|
||
}
|
||
)
|
||
return TraceDetail(
|
||
trace_id=trace_id,
|
||
character_ids=character_ids,
|
||
run_ids=run_ids,
|
||
session_ids=[],
|
||
items=items,
|
||
)
|
||
|
||
|
||
async def list_time_events(*, limit: int = 100, offset: int = 0) -> tuple[list[Event], int]:
|
||
"""列出辰正 time.* 事件（source='time_sense'），供起居注时间河展示（设计 §8 司命台）。"""
|
||
async with AsyncSessionLocal() as session:
|
||
result = await session.execute(
|
||
select(Event)
|
||
.where(Event.source == "time_sense")
|
||
.order_by(Event.created_at.desc())
|
||
.limit(limit)
|
||
.offset(offset)
|
||
)
|
||
items = list(result.scalars().all())
|
||
count_result = await session.execute(
|
||
select(func.count()).select_from(Event).where(Event.source == "time_sense")
|
||
)
|
||
total = int(count_result.scalar_one())
|
||
return items, total
|
||
|
||
|
||
async def list_traces(*, limit: int = 50) -> TraceListResponse:
|
||
"""列出最近活动的 trace_id 概览（跨 events / audit_events / idempotency_keys 聚合），
|
||
含每条链路涉及的人物与对话归属。"""
|
||
async with AsyncSessionLocal() as session:
|
||
merged: dict[str, dict[str, object]] = {}
|
||
|
||
def _entry(tid: str) -> dict[str, object]:
|
||
return merged.setdefault(
|
||
tid,
|
||
{
|
||
"trace_id": tid,
|
||
"last_seen": None,
|
||
"event_count": 0,
|
||
"audit_count": 0,
|
||
"idempotency_count": 0,
|
||
"character_ids": set(),
|
||
"session_ids": set(),
|
||
},
|
||
)
|
||
|
||
def _absorb(stats: list[tuple[str, int, object]], count_key: str) -> None:
|
||
for tid, cnt, last in stats:
|
||
entry = _entry(tid)
|
||
entry[count_key] = cnt
|
||
if last is not None and (entry["last_seen"] is None or last > entry["last_seen"]):
|
||
entry["last_seen"] = last
|
||
|
||
_absorb(await runtime_repo.summarize_by_trace(session), "event_count")
|
||
_absorb(await audit_repo.summarize_by_trace(session), "audit_count")
|
||
_absorb(await idempotency_repo.summarize_by_trace(session), "idempotency_count")
|
||
|
||
# 人物归属（events + audits 的 character_id）
|
||
for tid, cid in (
|
||
await session.execute(
|
||
select(Event.trace_id, Event.character_id).where(
|
||
Event.trace_id.is_not(None), Event.character_id.is_not(None)
|
||
)
|
||
)
|
||
).all():
|
||
_entry(tid)["character_ids"].add(cid)
|
||
for tid, cid in (
|
||
await session.execute(
|
||
select(AuditEvent.trace_id, AuditEvent.character_id).where(
|
||
AuditEvent.trace_id.is_not(None),
|
||
AuditEvent.character_id.is_not(None),
|
||
)
|
||
)
|
||
).all():
|
||
_entry(tid)["character_ids"].add(cid)
|
||
|
||
# 对话归属随线性会话模型退役；镜树归属随 M2c 司命台重设计接入（tree_id 聚合）
|
||
|
||
total = len(merged)
|
||
ordered = sorted(merged.values(), key=lambda e: e["last_seen"], reverse=True)
|
||
items = [
|
||
TraceSummary(
|
||
trace_id=str(entry["trace_id"]),
|
||
last_seen=entry["last_seen"], # type: ignore[arg-type]
|
||
character_ids=sorted(entry["character_ids"]), # type: ignore[arg-type]
|
||
session_ids=sorted(entry["session_ids"]), # type: ignore[arg-type]
|
||
event_count=int(entry["event_count"]),
|
||
audit_count=int(entry["audit_count"]),
|
||
idempotency_count=int(entry["idempotency_count"]),
|
||
)
|
||
for entry in ordered[:limit]
|
||
]
|
||
return TraceListResponse(items=items, total=total)
|