415 lines
15 KiB
Markdown
415 lines
15 KiB
Markdown
# agent-claw: Final Architecture Design
|
|
|
|
*Decisions locked 2026-05-04*
|
|
|
|
---
|
|
|
|
## System Overview
|
|
|
|
A serverless personal assistant running on AWS AgentCore. No always-on compute. Telegram-first, channel-agnostic by design. Two AgentCore runtimes: one user-facing, one background (REM).
|
|
|
|
```
|
|
Telegram ──→ API GW ──→ Lambda(tg-ingest) ──→ SQS FIFO ──→ Lambda(agent-runner)
|
|
│
|
|
InvokeAgentRuntime(Runtime 1)
|
|
│
|
|
Runtime 1: main assistant
|
|
reads: AgentCore Memory, factbase (Gateway),
|
|
S3 workspace
|
|
delivers via: channel adapter → Telegram
|
|
│
|
|
session ends (idle)
|
|
│
|
|
EventBridge ──→ Lambda(rem-runner)
|
|
│
|
|
InvokeAgentRuntime(Runtime 2)
|
|
│
|
|
Runtime 2: REM agent
|
|
calls: factbase workflow add/maintain
|
|
reads: AgentCore Memory session summary
|
|
writes: factbase (via AgentCore Gateway)
|
|
|
|
EventBridge ──→ Lambda(heartbeat-runner) ──→ InvokeAgentRuntime(Runtime 1, heartbeat prompt)
|
|
EventBridge ──→ Lambda(rem-runner) ──→ InvokeAgentRuntime(Runtime 2, nightly maintain)
|
|
```
|
|
|
|
---
|
|
|
|
## CDK Stack
|
|
|
|
**Stack: `AgentClawStack`**
|
|
|
|
All resources are create-or-use-existing via CDK context parameters or SSM lookups. Deploy is idempotent.
|
|
|
|
```typescript
|
|
// Context parameters (CDK --context flags or cdk.json)
|
|
// agentcore-memory-arn: existing memory ARN (optional, creates if absent)
|
|
// workspace-bucket-name: existing S3 bucket (optional, creates if absent)
|
|
// telegram-bot-token-secret-arn: existing secret (required before deploy)
|
|
// factbase-gateway-url: AgentCore Gateway endpoint for factbase tools
|
|
```
|
|
|
|
**Resources provisioned:**
|
|
|
|
| Resource | Type | Notes |
|
|
|---|---|---|
|
|
| `workspace-bucket` | S3 | SOUL.md, AGENTS.md, USER.md, HEARTBEAT.md |
|
|
| `session-store` | DynamoDB | actor_id → {session_id, created_at} |
|
|
| `personal-memory` | AgentCore Memory | SUMMARIZATION + USER_PREFERENCE strategies |
|
|
| `runtime-1` | AgentCore Runtime | main assistant, CodeZip, Sonnet |
|
|
| `runtime-2` | AgentCore Runtime | REM agent, CodeZip, Haiku |
|
|
| `message-queue` | SQS FIFO | MessageGroupId=actor_id, serializes per user |
|
|
| `rem-queue` | SQS | triggers REM after session idle |
|
|
| `tg-ingest` | Lambda | validates webhook, enqueues, sends typing |
|
|
| `agent-runner` | Lambda | SQS trigger, invokes Runtime 1, delivers response |
|
|
| `rem-runner` | Lambda | EventBridge + SQS trigger, invokes Runtime 2 |
|
|
| `webhook-api` | API GW HTTP | POST /telegram → tg-ingest |
|
|
| `heartbeat-rule` | EventBridge | every 30m → agent-runner (heartbeat prompt) |
|
|
| `rem-nightly` | EventBridge | nightly → rem-runner (factbase maintain) |
|
|
| `secrets` | Secrets Manager | bot-token, brave-api-key |
|
|
|
|
---
|
|
|
|
## Runtime 1: Main Assistant
|
|
|
|
**Language:** Python + Strands + bedrock-agentcore SDK
|
|
**Deploy:** CodeZip
|
|
**Model:** `us.anthropic.claude-3-7-sonnet-20250219-v1:0`
|
|
**Lifecycle:** `idleRuntimeSessionTimeout: 21600` (6hr), `maxLifetime: 28800` (8hr)
|
|
|
|
### Session start
|
|
1. Check in-memory cache for workspace files (SOUL.md, AGENTS.md, USER.md)
|
|
2. If cache miss: load from S3, cache in-memory
|
|
3. `RetrieveMemoryRecords(query=recent_message, namespace=/preferences/actor_id/)` → top-k preferences
|
|
4. `ListEvents(session_id)` → recent conversation turns (if resuming)
|
|
5. Build system prompt: workspace files + retrieved memory
|
|
6. Initialize Strands agent with `AgentCoreMemorySessionManager`
|
|
|
|
### Tools
|
|
|
|
| Tool | Description |
|
|
|---|---|
|
|
| `send_message(text, metadata?)` | Delivers via channel adapter (decoupled) |
|
|
| `web_search(query)` | Brave Search API |
|
|
| `web_fetch(url)` | HTTP + readability extraction |
|
|
| `read_workspace_file(path)` | S3 get_object |
|
|
| `write_workspace_file(path, content)` | S3 put_object + update in-memory cache |
|
|
| `search_memory(query)` | AgentCore Memory RetrieveMemoryRecords |
|
|
| `factbase_search(query)` | factbase via AgentCore Gateway |
|
|
|
|
**NOT included in Runtime 1:** factbase write tools (writes are async via Runtime 2 only)
|
|
|
|
### Channel adapter (decoupled)
|
|
|
|
```python
|
|
class ChannelAdapter(Protocol):
|
|
def send_message(self, text: str) -> str: ... # returns message_id
|
|
def send_typing(self) -> None: ...
|
|
def edit_message(self, msg_id: str, text: str) -> None: ...
|
|
|
|
class TelegramAdapter:
|
|
def send_message(self, text): ...
|
|
def send_typing(self): ... # sendChatAction(typing)
|
|
def edit_message(self, msg_id, text): ...
|
|
|
|
# Future: SlackAdapter, DiscordAdapter
|
|
# Channel type + target_id come from SQS message payload
|
|
# agent-runner instantiates correct adapter before invoking Runtime 1
|
|
# adapter is passed in payload, not hardcoded in agent
|
|
```
|
|
|
|
The agent calls `send_message(text)` — it doesn't know or care about Telegram specifics.
|
|
|
|
---
|
|
|
|
## Runtime 2: REM Agent
|
|
|
|
**Language:** Python + Strands + bedrock-agentcore SDK
|
|
**Deploy:** CodeZip
|
|
**Model:** `us.anthropic.claude-3-5-haiku-20241022-v1:0`
|
|
**Lifecycle:** short sessions, `maxLifetime: 1800` (30min)
|
|
|
|
### Trigger: session ended
|
|
1. agent-runner Lambda enqueues session summary to rem-queue when main session goes idle
|
|
2. rem-runner Lambda invokes Runtime 2 with:
|
|
- session summary (from AgentCore Memory SUMMARIZATION)
|
|
- actor_id for namespace routing
|
|
|
|
### Trigger: nightly (EventBridge cron)
|
|
1. rem-runner Lambda invokes Runtime 2 with: "Run factbase maintain"
|
|
|
|
### Tools
|
|
|
|
| Tool | Description |
|
|
|---|---|
|
|
| `factbase_workflow_add(summary)` | factbase workflow add via AgentCore Gateway |
|
|
| `factbase_workflow_maintain()` | factbase maintain workflow |
|
|
| `read_workspace_file(path)` | S3 (for HEARTBEAT.md context) |
|
|
| `write_workspace_file(path, content)` | S3 (update HEARTBEAT.md if needed) |
|
|
|
|
### REM flow
|
|
```
|
|
Receive: session summary text
|
|
→ factbase workflow add: "Here is a conversation summary. Extract any new or updated
|
|
facts about people, projects, or decisions and add/update factbase entities."
|
|
→ factbase guides agent through: what's new, what conflicts, what to write
|
|
→ On nightly schedule: factbase workflow maintain
|
|
```
|
|
|
|
---
|
|
|
|
## Lambda: tg-ingest
|
|
|
|
**Trigger:** API Gateway POST /telegram
|
|
**Runtime:** Python 3.12
|
|
|
|
```python
|
|
def handler(event, context):
|
|
# 1. Validate Telegram secret token header
|
|
if not validate_telegram_secret(event):
|
|
return {"statusCode": 403}
|
|
|
|
# 2. Parse Telegram Update
|
|
update = json.loads(event["body"])
|
|
chat_id = extract_chat_id(update)
|
|
message_text = extract_message(update)
|
|
|
|
# 3. Send typing action immediately (non-blocking)
|
|
send_telegram_typing(chat_id)
|
|
|
|
# 4. Enqueue to SQS FIFO
|
|
sqs.send_message(
|
|
QueueUrl=MESSAGE_QUEUE_URL,
|
|
MessageGroupId=str(chat_id), # serializes per conversation
|
|
MessageDeduplicationId=str(update["update_id"]),
|
|
MessageBody=json.dumps({
|
|
"channel": "telegram",
|
|
"chat_id": chat_id,
|
|
"messages": [{"text": message_text, "from": update["message"]["from"]}],
|
|
"update_id": update["update_id"],
|
|
"timestamp": update["message"]["date"],
|
|
})
|
|
)
|
|
|
|
return {"statusCode": 200} # Telegram acks within 3s requirement
|
|
```
|
|
|
|
---
|
|
|
|
## Lambda: agent-runner
|
|
|
|
**Trigger:** SQS FIFO (batch_size=10, MessageGroupId=actor_id)
|
|
**Runtime:** Python 3.12
|
|
|
|
```python
|
|
def handler(event, context):
|
|
# 1. Batch all SQS records (same actor, serialized by FIFO)
|
|
records = [json.loads(r["body"]) for r in event["Records"]]
|
|
channel = records[0]["channel"]
|
|
chat_id = records[0]["chat_id"]
|
|
actor_id = f"{channel}:{chat_id}"
|
|
|
|
# 2. Look up or create session in DynamoDB
|
|
session_id = get_or_create_session(actor_id)
|
|
|
|
# 3. Bundle messages
|
|
if len(records) == 1:
|
|
prompt = records[0]["messages"][0]["text"]
|
|
else:
|
|
msgs = "\n".join(f"[{i+1}] {r['messages'][0]['text']}" for i, r in enumerate(records))
|
|
prompt = f"You have {len(records)} queued messages:\n{msgs}"
|
|
|
|
# 4. Invoke Runtime 1
|
|
payload = {
|
|
"prompt": prompt,
|
|
"actor_id": actor_id,
|
|
"session_id": session_id,
|
|
"channel_adapter": {"type": channel, "target_id": str(chat_id)},
|
|
"bot_token_secret_arn": BOT_TOKEN_SECRET_ARN,
|
|
}
|
|
|
|
response = agentcore.invoke_agent_runtime(
|
|
agentRuntimeArn=RUNTIME_1_ARN,
|
|
runtimeSessionId=session_id,
|
|
payload=json.dumps(payload).encode()
|
|
)
|
|
|
|
# 5. Stream response — agent delivers its own messages via send_message tool
|
|
for chunk in response["response"]:
|
|
pass # consume stream; agent handles delivery internally
|
|
|
|
# 6. Enqueue session summary for REM processing
|
|
sqs.send_message(QueueUrl=REM_QUEUE_URL, MessageBody=json.dumps({
|
|
"actor_id": actor_id,
|
|
"session_id": session_id,
|
|
}))
|
|
```
|
|
|
|
---
|
|
|
|
## Lambda: rem-runner
|
|
|
|
**Trigger:** SQS (rem-queue) + EventBridge (nightly)
|
|
**Runtime:** Python 3.12
|
|
|
|
```python
|
|
def handler(event, context):
|
|
# Determine trigger source
|
|
if is_eventbridge(event):
|
|
# Nightly: run maintain on factbase
|
|
payload = {"prompt": "Run factbase maintain workflow to consolidate memories, detect conflicts, and generate review questions."}
|
|
session_id = f"rem-nightly-{today()}"
|
|
else:
|
|
# Post-session: extract new facts
|
|
record = json.loads(event["Records"][0]["body"])
|
|
actor_id = record["actor_id"]
|
|
session_id = record["session_id"]
|
|
summary = get_session_summary(actor_id, session_id)
|
|
payload = {
|
|
"prompt": f"Conversation summary to process:\n\n{summary}\n\nExtract new entities, facts, and updates.",
|
|
"actor_id": actor_id,
|
|
}
|
|
# Use dedicated REM session (separate from user session)
|
|
session_id = f"rem-{actor_id}-{today()}"
|
|
|
|
agentcore.invoke_agent_runtime(
|
|
agentRuntimeArn=RUNTIME_2_ARN,
|
|
runtimeSessionId=session_id,
|
|
payload=json.dumps(payload).encode()
|
|
)
|
|
```
|
|
|
|
---
|
|
|
|
## Factbase KB Design (perspective.yaml)
|
|
|
|
```yaml
|
|
organization: Daniel Levy
|
|
focus: Personal chief of staff KB — professional network, projects, commitments, decisions
|
|
|
|
entity_types:
|
|
person: Professional contacts — colleagues, customers, AWS team, interviewers, vendors
|
|
company: Organizations — AWS customers, target employers, partners, tools/services
|
|
project: Active and past projects — work deliverables, side projects, products
|
|
commitment: Explicit commitments made (by Daniel or to Daniel) with due dates
|
|
decision: Significant decisions made, rationale, outcomes
|
|
meeting: Key conversations — participants, decisions, follow-ups
|
|
|
|
review:
|
|
stale_days: 90
|
|
required_fields:
|
|
person: [current_role, company, last_contact_date]
|
|
commitment: [due_date, status, owner]
|
|
decision: [date, outcome]
|
|
```
|
|
|
|
---
|
|
|
|
## S3 Workspace Layout
|
|
|
|
```
|
|
s3://agent-claw-workspace-{account_id}/
|
|
├── SOUL.md ← personality, tone, operating rules
|
|
├── AGENTS.md ← workspace conventions, memory rules
|
|
├── USER.md ← Daniel's profile, preferences
|
|
├── IDENTITY.md ← agent name, emoji
|
|
├── HEARTBEAT.md ← periodic task checklist
|
|
└── TOOLS.md ← tool-specific notes
|
|
```
|
|
|
|
Seeded from existing OpenClaw workspace files. CDK deploy script copies them if bucket is new.
|
|
|
|
---
|
|
|
|
## Project Structure
|
|
|
|
```
|
|
agent-claw/
|
|
├── cdk/ ← CDK TypeScript stack
|
|
│ ├── bin/agent-claw.ts
|
|
│ ├── lib/agent-claw-stack.ts
|
|
│ └── lib/constructs/
|
|
│ ├── agentcore-runtime.ts
|
|
│ ├── memory.ts
|
|
│ └── eventbridge-rules.ts
|
|
├── src/
|
|
│ ├── runtime-1/ ← Main assistant (CodeZip deploy)
|
|
│ │ ├── main.py
|
|
│ │ ├── tools/
|
|
│ │ │ ├── __init__.py
|
|
│ │ │ ├── web.py ← web_search, web_fetch
|
|
│ │ │ ├── workspace.py ← read/write S3 workspace files
|
|
│ │ │ ├── memory.py ← search_memory (AgentCore Memory)
|
|
│ │ │ └── messaging.py ← send_message (channel adapter)
|
|
│ │ ├── channels/
|
|
│ │ │ ├── adapter.py ← ChannelAdapter protocol
|
|
│ │ │ └── telegram.py ← TelegramAdapter
|
|
│ │ ├── prompt_builder.py ← builds system prompt from workspace + memory
|
|
│ │ └── pyproject.toml
|
|
│ ├── runtime-2/ ← REM agent (CodeZip deploy)
|
|
│ │ ├── main.py
|
|
│ │ ├── tools/
|
|
│ │ │ ├── factbase.py ← factbase workflow tools via AgentCore Gateway
|
|
│ │ │ └── workspace.py ← shared workspace tools
|
|
│ │ └── pyproject.toml
|
|
│ └── lambdas/
|
|
│ ├── tg-ingest/
|
|
│ │ ├── handler.py
|
|
│ │ └── requirements.txt
|
|
│ ├── agent-runner/
|
|
│ │ ├── handler.py
|
|
│ │ └── requirements.txt
|
|
│ └── rem-runner/
|
|
│ ├── handler.py
|
|
│ └── requirements.txt
|
|
├── workspace/ ← seed files for S3 workspace
|
|
│ ├── SOUL.md
|
|
│ ├── AGENTS.md
|
|
│ ├── USER.md
|
|
│ └── HEARTBEAT.md
|
|
└── README.md
|
|
```
|
|
|
|
---
|
|
|
|
## Build Phases
|
|
|
|
### Phase 0 — Infrastructure spike (1-2 days)
|
|
- CDK stack: deploy API GW + tg-ingest Lambda + SQS + AgentCore Runtime 1 (minimal agent)
|
|
- Test: Telegram message → Lambda → SQS → Runtime 1 → response back to Telegram
|
|
- Confirm: cold start latency, session warm behavior, basic round-trip
|
|
|
|
### Phase 1 — Full main assistant (1 week)
|
|
- Runtime 1 with all tools: web_search, web_fetch, workspace read/write, send_message
|
|
- System prompt builder loading from S3
|
|
- AgentCore Memory: short-term turns + USER_PREFERENCE
|
|
- Channel adapter: TelegramAdapter with typing action
|
|
- DynamoDB session mapping
|
|
|
|
**Done when:** full conversation works with personality, web search, and memory across messages.
|
|
|
|
### Phase 2 — REM + factbase (1 week)
|
|
- Runtime 2 deployed with factbase workflow tools
|
|
- rem-runner Lambda + rem-queue
|
|
- Post-session trigger from agent-runner
|
|
- EventBridge nightly REM schedule
|
|
- factbase perspective.yaml defined, KB seeded
|
|
|
|
**Done when:** after a conversation, REM fires and writes relevant facts to factbase; next session the agent can search factbase for context.
|
|
|
|
### Phase 3 — Heartbeat + scheduling (3-4 days)
|
|
- EventBridge heartbeat rule (every 30m → Runtime 1 with heartbeat prompt)
|
|
- HEARTBEAT.md checklist loaded at heartbeat time
|
|
- HEARTBEAT_OK suppression
|
|
- Dynamic cron via `create_cron` tool (EventBridge SDK)
|
|
|
|
### Phase 4 — Polish
|
|
- Streaming progress (Telegram edit-in-place)
|
|
- Commitments system
|
|
- Error handling, DLQ, CloudWatch alarms
|
|
- Multi-user support (actor_id scoping already built in)
|
|
|
|
---
|
|
|
|
*Design finalized 2026-05-04. All pre-development questions resolved.*
|