import os import boto3 # Cache: built once per warm session (shared base only) _base_prompt: str | None = None def build_system_prompt(user_context: str = '') -> str: """Build system prompt from S3 workspace files + optional per-user context.""" base = _get_base_prompt() if user_context: return base + f'\n\n---\n\n## User\n{user_context}' return base def _get_base_prompt() -> str: global _base_prompt if _base_prompt is not None: return _base_prompt bucket = os.environ.get('WORKSPACE_BUCKET_NAME', '') or 'agent-claw-workspace-495395224548' print(f'[prompt_builder] Loading from bucket: {bucket!r}') if not bucket: print('[prompt_builder] WARNING: WORKSPACE_BUCKET_NAME not set!') _base_prompt = 'You are a helpful personal assistant.' return _base_prompt s3 = boto3.client('s3') parts = [] for fname in ['SOUL.md', 'AGENTS.md', 'IDENTITY.md', 'TOOLS.md']: try: obj = s3.get_object(Bucket=bucket, Key=fname) content = obj['Body'].read().decode('utf-8') parts.append(f'## {fname}\n{content}') print(f'[prompt_builder] Loaded {fname} ({len(content)} bytes)') except Exception as e: print(f'[prompt_builder] Failed to load {fname}: {e}') parts.append('## Runtime\nRuntime: agent-claw | host=AgentCore | model=bedrock-claude-sonnet | channel=telegram\nCurrent date/time is provided by the system. Timezone: America/Chicago.') _base_prompt = '\n\n---\n\n'.join(parts) print(f'[prompt_builder] Base prompt built: {len(_base_prompt)} chars') return _base_prompt def invalidate_prompt() -> None: """Force rebuild of system prompt on next invocation (call after workspace write).""" global _base_prompt _base_prompt = None