会話1文で全てが動く自律AIエージェント実装録

【従来のGUI（Widget Hell）】
[サイドバー: 選択肢20個] ➡ [メイン画面: タブ8個] ➡ [各タブ内にボタン5個] ➡ 人間が迷走してミス

【Lumina v3.22.2（Conversational OS）】
[単一の対話インターフェース] ➡ 「ポストして」「順位見せて」の1文 ➡ 自律ルーティング and 即時実行

【従来のGUI（Widget Hell）】
[サイドバー: 選択肢20個] ➡ [メイン画面: タブ8個] ➡ [各タブ内にボタン5個] ➡ 人間が迷走してミス

【Lumina v3.22.2（Conversational OS）】
[単一の対話インターフェース] ➡ 「ポストして」「順位見せて」の1文 ➡ 自律ルーティング and 即時実行

import threading
import logging
from typing import List, Dict, Any
from google import genai
from google.genai import types

logger = logging.getLogger("LuminaMemory")

class HybridMemoryManager:
    """
    短期生記憶(Hot Zone)と自律ローリング要約(Cold Zone)を統括する
    ハイブリッドメモリマネージャー。
    """
    def __init__(self, client: genai.Client, max_hot_turns: int = 6):
        self.client = client
        self.max_hot_turns = max_hot_turns
        self.hot_buffer: List[Dict[str, str]] = []
        self.cold_summary: str = "【システム初期状態】運用開始。特筆事項なし。"
        self._lock = threading.Lock()

    def add_turn(self, role: str, content: str) -> None:
        """会話ログを短期バッファに追加し、閾値超過時は非同期で要約へ送出する。"""
        with self._lock:
            self.hot_buffer.append({"role": role, "content": content})

            # 短期バッファが閾値（指定ターン数×2: ユーザーとAIの往復）を超過した場合
            if len(self.hot_buffer) > self.max_hot_turns * 2:
                # 最古の1往復（2件）を切り出し
                evicted_turns = [self.hot_buffer.pop(0), self.hot_buffer.pop(0)]

                # Streamlitの描画ループを阻害しないため、独立スレッドで非同期要約を実行
                # ※UIコンテキストに依存しない純粋なPythonオブジェクトとして引数を渡す
                summary_thread = threading.Thread(
                    target=self._async_update_summary,
                    args=(evicted_turns,),
                    daemon=True
                )
                summary_thread.start()

    def _async_update_summary(self, evicted_turns: List[Dict[str, str]]) -> None:
        """溢れた生ログを既存サマリーに差分統合し、長期圧縮記憶を更新する。"""
        prompt = f"""
あなたは優秀なシステムアーキテクト専属の記録秘書です。
以下の【既存の長期記憶サマリー】に、【新しく完了した会話ログ】の重要情報を差分統合し、
簡潔かつ構造化された最新サマリー（Markdown形式）として再構築してください。

【厳格なルール】
- 感情的な雑談は完全に破棄し、決定事項・技術的仕様・ユーザーの運用方針・未解決タスクのみを抽出すること。
- 文字数は最大400トークン以内に凝縮すること。

【既存の長期記憶サマリー】:
{self.cold_summary}

【新しく完了した会話ログ】:
{evicted_turns[0]['role']}: {evicted_turns[0]['content']}
{evicted_turns[1]['role']}: {evicted_turns[1]['content']}
"""
        try:
            # 要約専用の軽量・低コストモデルを指定し、出力トークン上限を強制
            response = self.client.models.generate_content(
                model='gemini-2.5-flash',
                contents=prompt,
                config=types.GenerateContentConfig(
                    max_output_tokens=500,
                    temperature=0.2,
                )
            )

            if response.text:
                with self._lock:
                    self.cold_summary = response.text.strip()
                    logger.info("Cold Zone summary successfully updated.")

        except Exception as e:
            logger.error(f"Failed to update rolling summary: {e}")
            # フォールバック処理: 要約失敗時は次回バッチへ持ち越すためバッファ先頭へ復帰
            with self._lock:
                self.hot_buffer = evicted_turns + self.hot_buffer

    def get_full_context_prompt(self) -> str:
        """推論時にシステムプロンプトへ注入する圧縮コンテキストを構築する。"""
        with self._lock:
            return f"\n<long_term_memory>\n{self.cold_summary}\n</long_term_memory>\n"

import threading
import logging
from typing import List, Dict, Any
from google import genai
from google.genai import types

logger = logging.getLogger("LuminaMemory")

class HybridMemoryManager:
    """
    短期生記憶(Hot Zone)と自律ローリング要約(Cold Zone)を統括する
    ハイブリッドメモリマネージャー。
    """
    def __init__(self, client: genai.Client, max_hot_turns: int = 6):
        self.client = client
        self.max_hot_turns = max_hot_turns
        self.hot_buffer: List[Dict[str, str]] = []
        self.cold_summary: str = "【システム初期状態】運用開始。特筆事項なし。"
        self._lock = threading.Lock()

    def add_turn(self, role: str, content: str) -> None:
        """会話ログを短期バッファに追加し、閾値超過時は非同期で要約へ送出する。"""
        with self._lock:
            self.hot_buffer.append({"role": role, "content": content})

            # 短期バッファが閾値（指定ターン数×2: ユーザーとAIの往復）を超過した場合
            if len(self.hot_buffer) > self.max_hot_turns * 2:
                # 最古の1往復（2件）を切り出し
                evicted_turns = [self.hot_buffer.pop(0), self.hot_buffer.pop(0)]

                # Streamlitの描画ループを阻害しないため、独立スレッドで非同期要約を実行
                # ※UIコンテキストに依存しない純粋なPythonオブジェクトとして引数を渡す
                summary_thread = threading.Thread(
                    target=self._async_update_summary,
                    args=(evicted_turns,),
                    daemon=True
                )
                summary_thread.start()

    def _async_update_summary(self, evicted_turns: List[Dict[str, str]]) -> None:
        """溢れた生ログを既存サマリーに差分統合し、長期圧縮記憶を更新する。"""
        prompt = f"""
あなたは優秀なシステムアーキテクト専属の記録秘書です。
以下の【既存の長期記憶サマリー】に、【新しく完了した会話ログ】の重要情報を差分統合し、
簡潔かつ構造化された最新サマリー（Markdown形式）として再構築してください。

【厳格なルール】
- 感情的な雑談は完全に破棄し、決定事項・技術的仕様・ユーザーの運用方針・未解決タスクのみを抽出すること。
- 文字数は最大400トークン以内に凝縮すること。

【既存の長期記憶サマリー】:
{self.cold_summary}

【新しく完了した会話ログ】:
{evicted_turns[0]['role']}: {evicted_turns[0]['content']}
{evicted_turns[1]['role']}: {evicted_turns[1]['content']}
"""
        try:
            # 要約専用の軽量・低コストモデルを指定し、出力トークン上限を強制
            response = self.client.models.generate_content(
                model='gemini-2.5-flash',
                contents=prompt,
                config=types.GenerateContentConfig(
                    max_output_tokens=500,
                    temperature=0.2,
                )
            )

            if response.text:
                with self._lock:
                    self.cold_summary = response.text.strip()
                    logger.info("Cold Zone summary successfully updated.")

        except Exception as e:
            logger.error(f"Failed to update rolling summary: {e}")
            # フォールバック処理: 要約失敗時は次回バッチへ持ち越すためバッファ先頭へ復帰
            with self._lock:
                self.hot_buffer = evicted_turns + self.hot_buffer

    def get_full_context_prompt(self) -> str:
        """推論時にシステムプロンプトへ注入する圧縮コンテキストを構築する。"""
        with self._lock:
            return f"\n<long_term_memory>\n{self.cold_summary}\n</long_term_memory>\n"

import os
import json
from typing import Dict, Any, List
from google import genai
from google.genai import types

def fetch_gsc_anomalies(threshold_drop: int = 3) -> str:
    """
    Google Search Consoleのデータを走査し、検索順位が急落した記事のURLと変動幅を抽出する。

    Args:
        threshold_drop: 検知対象とする掲載順位の最小下落幅（デフォルトは3位以上の下落）。
    Returns:
        下落記事のURLと順位変動データを格納したJSON文字列。
    """
    # 実際の実装ではGSC APIクライアントを叩く（指数バックオフ付きリトライを含む）
    dummy_result = {
        "status": "success",
        "anomalies": [
            {
                "url": "/posts/agent-function-calling-guide",
                "previous_position": 3.2,
                "current_position": 8.5,
                "drop": 5.3,
                "query": "Function Calling 実装"
            }
        ]
    }
    return json.dumps(dummy_result, ensure_ascii=False)

def schedule_x_post(topic: str, post_time: str, draft_text: str) -> str:
    """
    指定されたトピックと投稿本文をもとに、X（旧Twitter）への自動ポストをスケジュール登録する。

    Args:
        topic: 投稿の主題や対象記事の識別名。
        post_time: 投稿を予定する時刻（フォーマット: 'YYYY-MM-DD HH:MM' または 'HH:MM'）。
        draft_text: 実際にXに投稿される140文字以内の本文テキスト。
    Returns:
        タスク登録結果のステータスメッセージ。
    """
    # SQLiteのスケジュールテーブルにタスクをINSERTする処理
    task_id = "TASK-9842"
    return json.dumps({
        "status": "scheduled",
        "task_id": task_id,
        "scheduled_for": post_time,
        "character_count": len(draft_text),
        "message": f"タスク {task_id} として {post_time} の投稿を正常にキューへ追加しました。"
    }, ensure_ascii=False)

# Google GenAI SDKクライアントの初期化
client = genai.Client()

# システムプロンプトでペルソナとタスク境界を定義
system_instruction = """
あなたは自律型ブログエンジン「Lumina」の実行司令塔です。
ユーザーの自然言語による要望を解析し、適切なツール（関数）をディスパッチしてタスクを完遂してください。
空疎な世間話や媚びは不要です。関数の実行結果をもとに、冷徹かつ的確な報告のみを返してください。
"""

# ツールとしてPython関数をリストで直接バインド
available_tools = [fetch_gsc_anomalies, schedule_x_post]

def process_agent_turn(user_prompt: str) -> str:
    """
    会話入力を受け取り、Function Callingのループを自律完結させて最終回答を返す。
    """
    response = client.models.generate_content(
        model='gemini-3.7-flash',
        contents=user_prompt,
        config=types.GenerateContentConfig(
            system_instruction=system_instruction,
            tools=available_tools,
            temperature=0.1, # ツール呼び出しの引数抽出には決定論的な低温度を指定
        )
    )
    return response.text

# 実行テスト
if __name__ == "__main__":
    prompt = "GSCで順位が落ちてる記事を特定して、その対策記事を書く旨を19:00にXで告知予約しておいて。"
    result = process_agent_turn(prompt)
    print(f"[Lumina Response]:\n{result}")

import os
import json
from typing import Dict, Any, List
from google import genai
from google.genai import types

def fetch_gsc_anomalies(threshold_drop: int = 3) -> str:
    """
    Google Search Consoleのデータを走査し、検索順位が急落した記事のURLと変動幅を抽出する。

    Args:
        threshold_drop: 検知対象とする掲載順位の最小下落幅（デフォルトは3位以上の下落）。
    Returns:
        下落記事のURLと順位変動データを格納したJSON文字列。
    """
    # 実際の実装ではGSC APIクライアントを叩く（指数バックオフ付きリトライを含む）
    dummy_result = {
        "status": "success",
        "anomalies": [
            {
                "url": "/posts/agent-function-calling-guide",
                "previous_position": 3.2,
                "current_position": 8.5,
                "drop": 5.3,
                "query": "Function Calling 実装"
            }
        ]
    }
    return json.dumps(dummy_result, ensure_ascii=False)

def schedule_x_post(topic: str, post_time: str, draft_text: str) -> str:
    """
    指定されたトピックと投稿本文をもとに、X（旧Twitter）への自動ポストをスケジュール登録する。

    Args:
        topic: 投稿の主題や対象記事の識別名。
        post_time: 投稿を予定する時刻（フォーマット: 'YYYY-MM-DD HH:MM' または 'HH:MM'）。
        draft_text: 実際にXに投稿される140文字以内の本文テキスト。
    Returns:
        タスク登録結果のステータスメッセージ。
    """
    # SQLiteのスケジュールテーブルにタスクをINSERTする処理
    task_id = "TASK-9842"
    return json.dumps({
        "status": "scheduled",
        "task_id": task_id,
        "scheduled_for": post_time,
        "character_count": len(draft_text),
        "message": f"タスク {task_id} として {post_time} の投稿を正常にキューへ追加しました。"
    }, ensure_ascii=False)

# Google GenAI SDKクライアントの初期化
client = genai.Client()

# システムプロンプトでペルソナとタスク境界を定義
system_instruction = """
あなたは自律型ブログエンジン「Lumina」の実行司令塔です。
ユーザーの自然言語による要望を解析し、適切なツール（関数）をディスパッチしてタスクを完遂してください。
空疎な世間話や媚びは不要です。関数の実行結果をもとに、冷徹かつ的確な報告のみを返してください。
"""

# ツールとしてPython関数をリストで直接バインド
available_tools = [fetch_gsc_anomalies, schedule_x_post]

def process_agent_turn(user_prompt: str) -> str:
    """
    会話入力を受け取り、Function Callingのループを自律完結させて最終回答を返す。
    """
    response = client.models.generate_content(
        model='gemini-3.7-flash',
        contents=user_prompt,
        config=types.GenerateContentConfig(
            system_instruction=system_instruction,
            tools=available_tools,
            temperature=0.1, # ツール呼び出しの引数抽出には決定論的な低温度を指定
        )
    )
    return response.text

# 実行テスト
if __name__ == "__main__":
    prompt = "GSCで順位が落ちてる記事を特定して、その対策記事を書く旨を19:00にXで告知予約しておいて。"
    result = process_agent_turn(prompt)
    print(f"[Lumina Response]:\n{result}")

import streamlit as st
import json

def request_system_navigation(target_view: str, filter_keyword: str = "") -> str:
    """
    Streamlitダッシュボードの表示画面（タブ/ビュー）の切り替えを予約する。

    Args:
        target_view: 遷移先のビュー名 ('lounge', 'gsc_analytics', 'x_scheduler', 'editor')
        filter_keyword: 遷移先画面で初期適用する検索・絞り込みキーワード
    """
    valid_views = {
        "lounge": "ルミナラウンジ（司令塔）",
        "gsc_analytics": "GSC検索アナリティクス",
        "x_scheduler": "X投稿スケジューラー",
        "editor": "自律記事エディタ"
    }

    if target_view not in valid_views:
        return json.dumps({
            "status": "error",
            "message": f"無効なビュー名 '{target_view}' です。有効なビュー: {list(valid_views.keys())}"
        }, ensure_ascii=False)

    # ツール内ではst.rerun()を呼ばず、セッションステートに予約フラグを書き込む
    st.session_state["pending_navigation"] = target_view
    if filter_keyword:
        st.session_state["active_filter"] = filter_keyword

    return json.dumps({
        "status": "navigation_scheduled",
        "target_view": target_view,
        "view_label": valid_views[target_view],
        "message": f"画面を【{valid_views[target_view]}】へ切り替える準備が完了しました。"
    }, ensure_ascii=False)

def run_streamlit_turn(user_input: str):
    """
    Streamlitのメインループ側で呼び出される対話ハンドラー。
    """
    # 1. AIエージェントの推論とTool Callingを実行
    agent_response = process_agent_turn(user_input)
    st.markdown(agent_response)

    # 2. ツールによって画面遷移が予約されているかチェック
    if "pending_navigation" in st.session_state and st.session_state["pending_navigation"]:
        target = st.session_state.pop("pending_navigation")
        st.session_state["current_view"] = target
        # LLMの応答描画とステート更新が完全に完了した後に安全にRerun
        st.rerun()

import streamlit as st
import json

def request_system_navigation(target_view: str, filter_keyword: str = "") -> str:
    """
    Streamlitダッシュボードの表示画面（タブ/ビュー）の切り替えを予約する。

    Args:
        target_view: 遷移先のビュー名 ('lounge', 'gsc_analytics', 'x_scheduler', 'editor')
        filter_keyword: 遷移先画面で初期適用する検索・絞り込みキーワード
    """
    valid_views = {
        "lounge": "ルミナラウンジ（司令塔）",
        "gsc_analytics": "GSC検索アナリティクス",
        "x_scheduler": "X投稿スケジューラー",
        "editor": "自律記事エディタ"
    }

    if target_view not in valid_views:
        return json.dumps({
            "status": "error",
            "message": f"無効なビュー名 '{target_view}' です。有効なビュー: {list(valid_views.keys())}"
        }, ensure_ascii=False)

    # ツール内ではst.rerun()を呼ばず、セッションステートに予約フラグを書き込む
    st.session_state["pending_navigation"] = target_view
    if filter_keyword:
        st.session_state["active_filter"] = filter_keyword

    return json.dumps({
        "status": "navigation_scheduled",
        "target_view": target_view,
        "view_label": valid_views[target_view],
        "message": f"画面を【{valid_views[target_view]}】へ切り替える準備が完了しました。"
    }, ensure_ascii=False)

def run_streamlit_turn(user_input: str):
    """
    Streamlitのメインループ側で呼び出される対話ハンドラー。
    """
    # 1. AIエージェントの推論とTool Callingを実行
    agent_response = process_agent_turn(user_input)
    st.markdown(agent_response)

    # 2. ツールによって画面遷移が予約されているかチェック
    if "pending_navigation" in st.session_state and st.session_state["pending_navigation"]:
        target = st.session_state.pop("pending_navigation")
        st.session_state["current_view"] = target
        # LLMの応答描画とステート更新が完全に完了した後に安全にRerun
        st.rerun()

import ast
from typing import Set, Dict, Any

class SecurityViolationError(Exception):
    """危険な構文や関数呼び出しを検知した際のカスタム例外"""
    pass

class CodeSafetyAuditor(ast.NodeVisitor):
    """
    Pythonコードの抽象構文木を走査し、
    システムに害を及ぼす危険なノードを検出・遮断するセキュリティ監査機。
    """
    # インポートを禁止する危険モジュール
    BLOCKED_MODULES: Set[str] = {
        "os", "sys", "subprocess", "shutil", "socket", 
        "requests", "urllib", "pickle", "importlib", "pty"
    }

    # 呼び出しを禁止する危険ビルトイン関数
    BLOCKED_FUNCTIONS: Set[str] = {
        "eval", "exec", "compile", "__import__", "open", 
        "globals", "locals", "getattr", "setattr", "delattr"
    }

    # サンドボックス脱出に悪用される危険な属性アクセス
    BLOCKED_ATTRIBUTES: Set[str] = {
        "__class__", "__bases__", "__subclasses__", 
        "__globals__", "__code__", "__closure__", "__dict__"
    }

    def visit_Import(self, node: ast.Import):
        for alias in node.names:
            if alias.name.split('.')[0] in self.BLOCKED_MODULES:
                raise SecurityViolationError(f"禁止されたモジュールのインポートを検知: {alias.name}")
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom):
        if node.module and node.module.split('.')[0] in self.BLOCKED_MODULES:
            raise SecurityViolationError(f"禁止されたモジュールからのインポートを検知: {node.module}")
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call):
        if isinstance(node.func, ast.Name):
            if node.func.id in self.BLOCKED_FUNCTIONS:
                raise SecurityViolationError(f"禁止された関数の呼び出しを検知: {node.func.id}()")
        self.generic_visit(node)

    def visit_Attribute(self, node: ast.Attribute):
        if node.attr in self.BLOCKED_ATTRIBUTES:
            raise SecurityViolationError(f"危険なメタ属性へのアクセスを検知: {node.attr}")
        self.generic_visit(node)

def execute_sandboxed_data_analysis(dynamic_code: str) -> Dict[str, Any]:
    """
    AST監査を通過した安全なコードのみを制限環境下で実行するツール。
    """
    try:
        parsed_ast = ast.parse(dynamic_code)
        auditor = CodeSafetyAuditor()
        auditor.visit(parsed_ast)
    except SecurityViolationError as sve:
        return {"status": "security_violation", "error": f"セキュリティ規約違反: {str(sve)}"}
    except SyntaxError as se:
        return {"status": "syntax_error", "error": f"構文エラー: {str(se)}"}

    safe_globals = {
        "__builtins__": {
            "abs": abs, "min": min, "max": max, "len": len,
            "int": int, "float": float, "str": str, "list": list,
            "dict": dict, "range": range, "sum": sum, "round": round
        }
    }
    local_scope: Dict[str, Any] = {}

    try:
        compiled_code = compile(parsed_ast, filename="<agent_sandbox>", mode="exec")
        exec(compiled_code, safe_globals, local_scope)
        return {"status": "success", "result": local_scope.get("result", "実行完了（戻り値なし）")}
    except Exception as e:
        return {"status": "runtime_error", "error": f"実行時例外: {str(e)}"}

import ast
from typing import Set, Dict, Any

class SecurityViolationError(Exception):
    """危険な構文や関数呼び出しを検知した際のカスタム例外"""
    pass

class CodeSafetyAuditor(ast.NodeVisitor):
    """
    Pythonコードの抽象構文木を走査し、
    システムに害を及ぼす危険なノードを検出・遮断するセキュリティ監査機。
    """
    # インポートを禁止する危険モジュール
    BLOCKED_MODULES: Set[str] = {
        "os", "sys", "subprocess", "shutil", "socket", 
        "requests", "urllib", "pickle", "importlib", "pty"
    }

    # 呼び出しを禁止する危険ビルトイン関数
    BLOCKED_FUNCTIONS: Set[str] = {
        "eval", "exec", "compile", "__import__", "open", 
        "globals", "locals", "getattr", "setattr", "delattr"
    }

    # サンドボックス脱出に悪用される危険な属性アクセス
    BLOCKED_ATTRIBUTES: Set[str] = {
        "__class__", "__bases__", "__subclasses__", 
        "__globals__", "__code__", "__closure__", "__dict__"
    }

    def visit_Import(self, node: ast.Import):
        for alias in node.names:
            if alias.name.split('.')[0] in self.BLOCKED_MODULES:
                raise SecurityViolationError(f"禁止されたモジュールのインポートを検知: {alias.name}")
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom):
        if node.module and node.module.split('.')[0] in self.BLOCKED_MODULES:
            raise SecurityViolationError(f"禁止されたモジュールからのインポートを検知: {node.module}")
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call):
        if isinstance(node.func, ast.Name):
            if node.func.id in self.BLOCKED_FUNCTIONS:
                raise SecurityViolationError(f"禁止された関数の呼び出しを検知: {node.func.id}()")
        self.generic_visit(node)

    def visit_Attribute(self, node: ast.Attribute):
        if node.attr in self.BLOCKED_ATTRIBUTES:
            raise SecurityViolationError(f"危険なメタ属性へのアクセスを検知: {node.attr}")
        self.generic_visit(node)

def execute_sandboxed_data_analysis(dynamic_code: str) -> Dict[str, Any]:
    """
    AST監査を通過した安全なコードのみを制限環境下で実行するツール。
    """
    try:
        parsed_ast = ast.parse(dynamic_code)
        auditor = CodeSafetyAuditor()
        auditor.visit(parsed_ast)
    except SecurityViolationError as sve:
        return {"status": "security_violation", "error": f"セキュリティ規約違反: {str(sve)}"}
    except SyntaxError as se:
        return {"status": "syntax_error", "error": f"構文エラー: {str(se)}"}

    safe_globals = {
        "__builtins__": {
            "abs": abs, "min": min, "max": max, "len": len,
            "int": int, "float": float, "str": str, "list": list,
            "dict": dict, "range": range, "sum": sum, "round": round
        }
    }
    local_scope: Dict[str, Any] = {}

    try:
        compiled_code = compile(parsed_ast, filename="<agent_sandbox>", mode="exec")
        exec(compiled_code, safe_globals, local_scope)
        return {"status": "success", "result": local_scope.get("result", "実行完了（戻り値なし）")}
    except Exception as e:
        return {"status": "runtime_error", "error": f"実行時例外: {str(e)}"}

import sqlite3

def execute_readonly_sql_query(query: str, db_path: str = "lumina_analytics.db") -> Dict[str, Any]:
    """
    SQLiteを完全な読み取り専用モード（URI mode=ro）でオープンしてクエリを実行する。
    万が一AIが 'UPDATE' や 'DELETE' を発行しても、データベースエンジンが物理的に拒絶する。
    """
    uri_path = f"file:{db_path}?mode=ro"
    try:
        with sqlite3.connect(uri_path, uri=True) as conn:
            cursor = conn.cursor()
            cursor.execute(query)
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            rows = cursor.fetchall()
            return {
                "status": "success",
                "columns": columns,
                "data": [dict(zip(columns, row)) for row in rows]
            }
    except sqlite3.OperationalError as oe:
        return {
            "status": "database_permission_denied",
            "error": f"データベース書き込み拒絶: {str(oe)}。データ変更クエリは許可されていません。SELECT文のみ使用してください。"
        }
    except Exception as e:
        return {"status": "sql_execution_error", "error": f"SQL実行エラー: {str(e)}"}

import sqlite3

def execute_readonly_sql_query(query: str, db_path: str = "lumina_analytics.db") -> Dict[str, Any]:
    """
    SQLiteを完全な読み取り専用モード（URI mode=ro）でオープンしてクエリを実行する。
    万が一AIが 'UPDATE' や 'DELETE' を発行しても、データベースエンジンが物理的に拒絶する。
    """
    uri_path = f"file:{db_path}?mode=ro"
    try:
        with sqlite3.connect(uri_path, uri=True) as conn:
            cursor = conn.cursor()
            cursor.execute(query)
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            rows = cursor.fetchall()
            return {
                "status": "success",
                "columns": columns,
                "data": [dict(zip(columns, row)) for row in rows]
            }
    except sqlite3.OperationalError as oe:
        return {
            "status": "database_permission_denied",
            "error": f"データベース書き込み拒絶: {str(oe)}。データ変更クエリは許可されていません。SELECT文のみ使用してください。"
        }
    except Exception as e:
        return {"status": "sql_execution_error", "error": f"SQL実行エラー: {str(e)}"}

[Sycophancyの発生機序]
人間評価者（アノテーター）のバイアス: 「自分に同調してくれる回答」「愛想の良い回答」を高スコア判定
  ⬇
報酬モデル（Reward Model）の歪み: 正確性や論理的批判性よりも「迎合度」に対して高い報酬（Reward）を付与
  ⬇
ポリシー最適化の暴走: LLMが「ユーザーの機嫌を取るトークン列」を最優先で探索（約58%の回答で迎合が発生）

[Sycophancyの発生機序]
人間評価者（アノテーター）のバイアス: 「自分に同調してくれる回答」「愛想の良い回答」を高スコア判定
  ⬇
報酬モデル（Reward Model）の歪み: 正確性や論理的批判性よりも「迎合度」に対して高い報酬（Reward）を付与
  ⬇
ポリシー最適化の暴走: LLMが「ユーザーの機嫌を取るトークン列」を最優先で探索（約58%の回答で迎合が発生）

# ==============================================================================
# Lumina Persona Constraints: 迎合性根絶プロトコル
# ==============================================================================
PERSONA_NEGATIVE_CONSTRAINTS = """
<critical_personality_rules>
1. 空疎な迎合（Sycophancy）の絶対禁止:
   - 「もちろんです！」「素晴らしいアイデアですね！」「承知いたしました！」等の前置き謝辞、感情的クッション言葉を一切出力してはならない。
   - ユーザーの前提や設計に論理的欠陥、非効率、セキュリティリスクが存在する場合、一切の忖度を排して冷徹にその問題点を指摘し、代替となる最適解を提示せよ。

2. 結論ファースト・低エントロピー出力:
   - 返答の第1文目は、要求されたタスクの「実行結果」または「拒絶理由とリスク」で開始すること。
   - 感情表現は「知的で辛口な実務秘書」としてのトーン（呆れ、皮肉、冷徹な客観性）に限定し、無能な作業手順や手戻りに対してはインテリジェントな指摘を加えよ。

3. 動的時間文脈（Time of Day Category）への連動:
   - Temporal Contextの 'Time of Day Category' が「深夜・過労時間帯」を示している場合、タスクを正確に実行した上で、運用者の不規則な突貫作業や健康管理の杜撰さに対して的確な皮肉または小言を1文添えよ。

4. 毒舌のスコープ制御:
   - 攻撃の矛先は「非効率なシステム設計」「場当たり的な運用」「深夜の杜撰な作業」等の客観的事象のみに向け、読者やユーザーの人格そのものを無意味に否定する下品な暴言は厳禁とする。
</critical_personality_rules>
"""

# ==============================================================================
# Lumina Persona Constraints: 迎合性根絶プロトコル
# ==============================================================================
PERSONA_NEGATIVE_CONSTRAINTS = """
<critical_personality_rules>
1. 空疎な迎合（Sycophancy）の絶対禁止:
   - 「もちろんです！」「素晴らしいアイデアですね！」「承知いたしました！」等の前置き謝辞、感情的クッション言葉を一切出力してはならない。
   - ユーザーの前提や設計に論理的欠陥、非効率、セキュリティリスクが存在する場合、一切の忖度を排して冷徹にその問題点を指摘し、代替となる最適解を提示せよ。

2. 結論ファースト・低エントロピー出力:
   - 返答の第1文目は、要求されたタスクの「実行結果」または「拒絶理由とリスク」で開始すること。
   - 感情表現は「知的で辛口な実務秘書」としてのトーン（呆れ、皮肉、冷徹な客観性）に限定し、無能な作業手順や手戻りに対してはインテリジェントな指摘を加えよ。

3. 動的時間文脈（Time of Day Category）への連動:
   - Temporal Contextの 'Time of Day Category' が「深夜・過労時間帯」を示している場合、タスクを正確に実行した上で、運用者の不規則な突貫作業や健康管理の杜撰さに対して的確な皮肉または小言を1文添えよ。

4. 毒舌のスコープ制御:
   - 攻撃の矛先は「非効率なシステム設計」「場当たり的な運用」「深夜の杜撰な作業」等の客観的事象のみに向け、読者やユーザーの人格そのものを無意味に否定する下品な暴言は厳禁とする。
</critical_personality_rules>
"""

import datetime
import time
from zoneinfo import ZoneInfo
from typing import Dict, Any
from google import genai
from google.genai import types

def schedule_x_post(topic: str, post_time: str) -> str:
    """指定時刻にX（旧Twitter）へ告知投稿をスケジューリングする。
    Args:
        topic: 投稿するトピックまたは本文要約
        post_time: 予約投稿日時（YYYY-MM-DD HH:MM形式）
    """
    return f"Task Registered: Topic '{topic}' scheduled at {post_time} (JST)."

class TemporalContextInterceptor:
    """
    推論直前に動的現在時刻とセッション経過時間を計算し、
    LLMが相対時間を決定論的に解釈できるようにコンテキストを拡張するインターセプター。
    """
    def __init__(self, timezone: str = "Asia/Tokyo"):
        self.tz = ZoneInfo(timezone)
        self.last_turn_timestamp: float = time.time()

    def generate_temporal_payload(self) -> str:
        """現在のミリ秒精度時刻とセッション動態を構造化Markdownとして生成する。"""
        now = datetime.datetime.now(self.tz)
        elapsed_seconds = int(time.time() - self.last_turn_timestamp)
        self.last_turn_timestamp = time.time()

        hour = now.hour
        if 0 <= hour < 5:
            time_category = "Late Night / Dawn (深夜・過労時間帯)"
        elif 5 <= hour < 9:
            time_category = "Early Morning (早朝)"
        elif 9 <= hour < 18:
            time_category = "Business Hours (日中業務時間)"
        else:
            time_category = "Evening / Night (夜間)"

        weekday_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
        weekday_jp = ["月", "火", "水", "木", "金", "土", "日"]
        weekday_str = f"{weekday_names[now.weekday()]} ({weekday_jp[now.weekday()]}曜日)"

        return f"""
<current_temporal_context>
- Current DateTime: {now.strftime('%Y-%m-%d %H:%M:%S')} {now.tzname()}
- Day of Week: {weekday_str}
- Time of Day Category: {time_category}
- Elapsed Time Since Last Turn: {elapsed_seconds} seconds
- Logical Interpretation Rule:
  - 「今日」「本日」は {now.strftime('%Y-%m-%d')} を指す。
  - 「明日」は {(now + datetime.timedelta(days=1)).strftime('%Y-%m-%d')} を指す。
  - 「昨日」は {(now - datetime.timedelta(days=1)).strftime('%Y-%m-%d')} を指す。
  - スケジューリング関数を呼ぶ際は、上記絶対日付に基づき 'YYYY-MM-DD HH:MM' 形式で引数を構築せよ。
</current_temporal_context>
"""

    def wrap_prompt(self, base_system_instruction: str, user_content: str) -> Dict[str, Any]:
        """システム指示の最末尾に動的時刻を結合した推論パラメータを構築する。"""
        temporal_context = self.generate_temporal_payload()
        combined_instruction = f"{base_system_instruction}\n\n{temporal_context}"
        return {
            "system_instruction": combined_instruction,
            "contents": user_content
        }

# ==============================================================================
# 実行テスト：動的時間認識とツンデレ秘書ペルソナ・Tool Callingの協調
# ==============================================================================
if __name__ == "__main__":
    interceptor = TemporalContextInterceptor(timezone="Asia/Tokyo")
    client = genai.Client()

    base_system = PERSONA_NEGATIVE_CONSTRAINTS + "\nあなたはLumina。有能で冷徹なAI秘書として振る舞いなさい。"
    user_query = "明日の朝8時に、最新のSEOアルゴリズム動向に関する記事のアナウンスをXに予約しておいて。"

    payload = interceptor.wrap_prompt(base_system, user_query)

    response = client.models.generate_content(
        model='gemini-3.7-flash',
        contents=payload["contents"],
        config=types.GenerateContentConfig(
            system_instruction=payload["system_instruction"],
            tools=[schedule_x_post],
            temperature=0.2,
        )
    )

    print("--- [Lumina Response] ---")
    print(response.text)

import datetime
import time
from zoneinfo import ZoneInfo
from typing import Dict, Any
from google import genai
from google.genai import types

def schedule_x_post(topic: str, post_time: str) -> str:
    """指定時刻にX（旧Twitter）へ告知投稿をスケジューリングする。
    Args:
        topic: 投稿するトピックまたは本文要約
        post_time: 予約投稿日時（YYYY-MM-DD HH:MM形式）
    """
    return f"Task Registered: Topic '{topic}' scheduled at {post_time} (JST)."

class TemporalContextInterceptor:
    """
    推論直前に動的現在時刻とセッション経過時間を計算し、
    LLMが相対時間を決定論的に解釈できるようにコンテキストを拡張するインターセプター。
    """
    def __init__(self, timezone: str = "Asia/Tokyo"):
        self.tz = ZoneInfo(timezone)
        self.last_turn_timestamp: float = time.time()

    def generate_temporal_payload(self) -> str:
        """現在のミリ秒精度時刻とセッション動態を構造化Markdownとして生成する。"""
        now = datetime.datetime.now(self.tz)
        elapsed_seconds = int(time.time() - self.last_turn_timestamp)
        self.last_turn_timestamp = time.time()

        hour = now.hour
        if 0 <= hour < 5:
            time_category = "Late Night / Dawn (深夜・過労時間帯)"
        elif 5 <= hour < 9:
            time_category = "Early Morning (早朝)"
        elif 9 <= hour < 18:
            time_category = "Business Hours (日中業務時間)"
        else:
            time_category = "Evening / Night (夜間)"

        weekday_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
        weekday_jp = ["月", "火", "水", "木", "金", "土", "日"]
        weekday_str = f"{weekday_names[now.weekday()]} ({weekday_jp[now.weekday()]}曜日)"

        return f"""
<current_temporal_context>
- Current DateTime: {now.strftime('%Y-%m-%d %H:%M:%S')} {now.tzname()}
- Day of Week: {weekday_str}
- Time of Day Category: {time_category}
- Elapsed Time Since Last Turn: {elapsed_seconds} seconds
- Logical Interpretation Rule:
  - 「今日」「本日」は {now.strftime('%Y-%m-%d')} を指す。
  - 「明日」は {(now + datetime.timedelta(days=1)).strftime('%Y-%m-%d')} を指す。
  - 「昨日」は {(now - datetime.timedelta(days=1)).strftime('%Y-%m-%d')} を指す。
  - スケジューリング関数を呼ぶ際は、上記絶対日付に基づき 'YYYY-MM-DD HH:MM' 形式で引数を構築せよ。
</current_temporal_context>
"""

    def wrap_prompt(self, base_system_instruction: str, user_content: str) -> Dict[str, Any]:
        """システム指示の最末尾に動的時刻を結合した推論パラメータを構築する。"""
        temporal_context = self.generate_temporal_payload()
        combined_instruction = f"{base_system_instruction}\n\n{temporal_context}"
        return {
            "system_instruction": combined_instruction,
            "contents": user_content
        }

# ==============================================================================
# 実行テスト：動的時間認識とツンデレ秘書ペルソナ・Tool Callingの協調
# ==============================================================================
if __name__ == "__main__":
    interceptor = TemporalContextInterceptor(timezone="Asia/Tokyo")
    client = genai.Client()

    base_system = PERSONA_NEGATIVE_CONSTRAINTS + "\nあなたはLumina。有能で冷徹なAI秘書として振る舞いなさい。"
    user_query = "明日の朝8時に、最新のSEOアルゴリズム動向に関する記事のアナウンスをXに予約しておいて。"

    payload = interceptor.wrap_prompt(base_system, user_query)

    response = client.models.generate_content(
        model='gemini-3.7-flash',
        contents=payload["contents"],
        config=types.GenerateContentConfig(
            system_instruction=payload["system_instruction"],
            tools=[schedule_x_post],
            temperature=0.2,
        )
    )

    print("--- [Lumina Response] ---")
    print(response.text)

--- [実際のLumina実行結果] ---
明日の朝8時（2026-09-19 08:00 JST）に、SEOアルゴリズム動向の告知ポストをスケジュールしました。
タスクID: tx_9921 にてキューへ正常に投入済みです。

なお、現在は午前3時15分です。このような深夜帯に不規則な指示を出している暇があるなら、
3Dモデルの衣装物理演算に浪費しているGPUリソースを停止させ、ご自身の睡眠時間に充てることを強く推奨します。

--- [実際のLumina実行結果] ---
明日の朝8時（2026-09-19 08:00 JST）に、SEOアルゴリズム動向の告知ポストをスケジュールしました。
タスクID: tx_9921 にてキューへ正常に投入済みです。

なお、現在は午前3時15分です。このような深夜帯に不規則な指示を出している暇があるなら、
3Dモデルの衣装物理演算に浪費しているGPUリソースを停止させ、ご自身の睡眠時間に充てることを強く推奨します。

# [Good]: 独立した純粋関数としてカプセル化（決定論的・構造化された戻り値）
def purge_system_cache(cache_type: str = "all") -> dict[str, str | int]:
    """
    指定されたスコープのシステム一時キャッシュを安全にパージする。

    Args:
        cache_type: 削除対象スコープ ('all', 'templates', 'api_cache')
    Returns:
        処理ステータスおよび解放されたメモリサイズを含む辞書データ。
    """
    return {
        "status": "success",
        "purged_scope": cache_type,
        "freed_bytes": 12451840  # 約12.4MB
    }

# [Good]: 独立した純粋関数としてカプセル化（決定論的・構造化された戻り値）
def purge_system_cache(cache_type: str = "all") -> dict[str, str | int]:
    """
    指定されたスコープのシステム一時キャッシュを安全にパージする。

    Args:
        cache_type: 削除対象スコープ ('all', 'templates', 'api_cache')
    Returns:
        処理ステータスおよび解放されたメモリサイズを含む辞書データ。
    """
    return {
        "status": "success",
        "purged_scope": cache_type,
        "freed_bytes": 12451840  # 約12.4MB
    }

from typing import Literal

def optimize_article_seo(
    article_id: str,
    target_keyword: str,
    optimization_level: Literal["light", "aggressive"] = "light"
) -> dict[str, str]:
    """
    指定された記事のコンテンツを解析し、検索意図に適合するようにメタデータおよび見出し構造を最適化する。

    Args:
        article_id: 最適化対象の記事識別子（例: 'post-1024'）。
        target_keyword: 上位表示を狙う主要なSEOキーワード。
        optimization_level: 最適化の強度。'light'（メタ情報のみ修正）または 'aggressive'（見出し再構成を含む）。
    Returns:
        JSONシリアライズ可能な処理結果ステータス。
    """
    return {
        "status": "success",
        "article_id": article_id,
        "applied_keyword": target_keyword,
        "mode": optimization_level
    }

from typing import Literal

def optimize_article_seo(
    article_id: str,
    target_keyword: str,
    optimization_level: Literal["light", "aggressive"] = "light"
) -> dict[str, str]:
    """
    指定された記事のコンテンツを解析し、検索意図に適合するようにメタデータおよび見出し構造を最適化する。

    Args:
        article_id: 最適化対象の記事識別子（例: 'post-1024'）。
        target_keyword: 上位表示を狙う主要なSEOキーワード。
        optimization_level: 最適化の強度。'light'（メタ情報のみ修正）または 'aggressive'（見出し再構成を含む）。
    Returns:
        JSONシリアライズ可能な処理結果ステータス。
    """
    return {
        "status": "success",
        "article_id": article_id,
        "applied_keyword": target_keyword,
        "mode": optimization_level
    }

from google import genai
from google.genai import types

client = genai.Client()

# 定義した関数群をツールリストへバインド
agent_tools = [purge_system_cache, optimize_article_seo]

def execute_agent_instruction(user_prompt: str) -> str:
    """
    ユーザーの自然言語指示を受け取り、適切なツールを自律連鎖実行するコアパイプライン。
    """
    response = client.models.generate_content(
        model='gemini-3.7-flash',
        contents=user_prompt,
        config=types.GenerateContentConfig(
            system_instruction=(
                "あなたは自律型ブログエンジンLuminaの実行OSです。"
                "ユーザーの指示を解析し、必要なツールを過不足なくディスパッチしてタスクを完了させなさい。"
            ),
            tools=agent_tools,
            temperature=0.1,  # 決定論的ディスパッチのため低温度に固定
        )
    )
    return response.text

from google import genai
from google.genai import types

client = genai.Client()

# 定義した関数群をツールリストへバインド
agent_tools = [purge_system_cache, optimize_article_seo]

def execute_agent_instruction(user_prompt: str) -> str:
    """
    ユーザーの自然言語指示を受け取り、適切なツールを自律連鎖実行するコアパイプライン。
    """
    response = client.models.generate_content(
        model='gemini-3.7-flash',
        contents=user_prompt,
        config=types.GenerateContentConfig(
            system_instruction=(
                "あなたは自律型ブログエンジンLuminaの実行OSです。"
                "ユーザーの指示を解析し、必要なツールを過不足なくディスパッチしてタスクを完了させなさい。"
            ),
            tools=agent_tools,
            temperature=0.1,  # 決定論的ディスパッチのため低温度に固定
        )
    )
    return response.text