openclaw-intelligence-core-public/syncpatch/meta_controller.py

from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Any

from tool_graph import build_tool_graph
from uncertainty_model import estimate_uncertainty
from bandit_policy import load_policy_candidate, apply_bandit_bias



def shadow_decision(message: str, analysis: dict[str, Any], family_candidates: list[dict[str, Any]] | None, tool_registry: dict[str, dict[str, Any]]) -> dict[str, Any]:
    graph = build_tool_graph(tool_registry)
    uncertainty = estimate_uncertainty(message, analysis, family_candidates)
    tools = list(analysis.get('tools') or [])
    families = [str((item or {}).get('family') or '') for item in (family_candidates or []) if (item or {}).get('family')]

    decision = 'answer_direct'
    reason = 'single_grounded_or_low_uncertainty'
    suggested_memory_mode = ''

    if uncertainty['level'] == 'high' and 'ambiguous_access' in families:
        decision = 'ask_clarification'
        reason = 'ambiguous_service_access'
    elif analysis.get('needs_memory') and analysis.get('needs_setup_context'):
        decision = 'run_plan'
        reason = 'mixed_memory_plus_setup'
        suggested_memory_mode = 'setup'
    elif analysis.get('needs_memory'):
        decision = 'use_memory_mode'
        reason = 'memory_required'
        suggested_memory_mode = 'profile' if analysis.get('task_type') == 'memory' else 'preference'
    elif analysis.get('needs_setup_context') or len(tools) > 1:
        decision = 'run_plan'
        reason = 'evidence_required'
    elif uncertainty['level'] == 'medium' and graph.get(tools[0], None) and graph[tools[0]].groundedness == 'weak':
        decision = 'run_plan'
        reason = 'weak_grounding_under_uncertainty'

    chosen_plan = str(analysis.get('composition_reason') or 'single_tool')
    policy = load_policy_candidate()
    bandit = apply_bandit_bias(
        base_decision=decision,
        base_reason=reason,
        chosen_plan=chosen_plan,
        families=families,
        uncertainty=uncertainty,
        policy=policy,
    )
    decision = bandit.get('decision', decision)
    reason = bandit.get('reason', reason)

    return {
        'ts': datetime.now(timezone.utc).isoformat(),
        'message': message,
        'decision': decision,
        'reason': reason,
        'suggested_memory_mode': suggested_memory_mode,
        'suggested_tools': tools,
        'uncertainty': uncertainty,
        'family_candidates': families,
        'normalized_task': f"{analysis.get('role','')}:{analysis.get('task_type','')}",
        'chosen_plan': chosen_plan,
        'policy_hint': {
            'plan_prior': bandit.get('plan_prior', {}),
            'family_priors': bandit.get('family_priors', []),
        },
    }



def log_shadow_decision(log_path, decision_row: dict[str, Any]) -> None:
    try:
        log_path.parent.mkdir(parents=True, exist_ok=True)
        with log_path.open('a', encoding='utf-8') as f:
            f.write(json.dumps(decision_row, ensure_ascii=False) + '\n')
    except Exception:
        pass