adam/exec-aide
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
This repository has been archived on 2026-08-04. You can view files and clone it, but cannot push or open issues or pull requests.
exec-aide/PLAN.md
Adam Moussa 2edac81cce Add implementation plan for ultraplan refinement
2026-04-30 16:49:04 -04:00

11 KiB
Raw Blame History

Exec Aide — Implementation Plan

Context

Adam wants a personal AI agent that monitors his Gmail inbox, classifies emails by urgency, detects bypassed work-order routing and unanswered threads, and delivers real-time Slack alerts plus a daily digest. The brief proposed Pipedream first, but that conflicts with Sea Haven conventions (SAM default, Secrets Manager, CloudFormation-managed resources). This plan goes straight to SAM.

Stack: exec-aide

Repo: exec-aide in Sea-Haven-Industries (private) Runtime: Python 3.12 / arm64 IaC: SAM (template.yaml) AI: Bedrock — Claude Haiku (us.anthropic.claude-haiku-4-5-20251001)

Directory Structure

exec-aide/
├── template.yaml
├── samconfig.toml.example
├── .gitignore
├── README.md
├── scripts/
│   └── get_gmail_token.py        # One-time OAuth token exchange
└── src/
    ├── requirements.txt           # Shared: google-api-python-client, google-auth, requests
    ├── fetch_classify/
    │   ├── __init__.py
    │   └── app.py                 # 15-min poll handler
    ├── daily_digest/
    │   ├── __init__.py
    │   └── app.py                 # 5 PM ET digest handler
    └── shared/
        ├── __init__.py
        ├── secrets.py             # Secrets Manager + SSM caching
        ├── gmail.py               # Gmail API: OAuth refresh, history sync, message fetch
        ├── classify.py            # Bedrock Haiku classification prompt
        ├── slack.py               # Slack DM posting, Block Kit builders
        └── dynamo.py              # DynamoDB state operations

Both functions share CodeUri: src/ with different handlers. This deviates slightly from the canonical per-function directory layout but avoids duplicating the shared module.

CloudFormation Resources

DynamoDB Table: exec-aide

Single table, composite key. Three item types distinguished by pk prefix.

Item pk sk Key attributes
Message MSG#<messageId> MSG thread_id, from_email, from_name, to_emails, cc_emails, subject, snippet, internal_date, classification (HIGH/NORMAL/LOW), classification_reason, bypassed_wo (bool), notified (bool), classified_date (YYYY-MM-DD)
Thread THD#<threadId> THD last_message_from, last_message_date, adam_last_reply_date, subject, unanswered_since
Sync meta META#sync META history_id, last_sync_at, initial_sync_done
  • GSI by-date: PK=classified_date, SK=sk — for daily digest date queries
  • TTL: ttl attribute, 90 days on MSG/THD items, no TTL on META
  • Billing: PAY_PER_REQUEST

Lambda Functions

Function Trigger Timeout Purpose
exec-aide-fetch-classify EventBridge rate(15 minutes) 120s Poll Gmail, classify, alert on HIGH
exec-aide-daily-digest EventBridge Scheduler cron(0 17 ? * MON-FRI *) tz=America/New_York 120s Build and send daily digest DM

Both get explicit CloudWatch log groups with 60-day retention.

EventBridge Scheduler (digest)

AWS::Scheduler::Schedule with ScheduleExpressionTimezone: America/New_York — handles DST correctly. Requires an IAM role granting lambda:InvokeFunction to the scheduler service.

IAM Permissions (per function)

fetch-classify:

  • DynamoDBCrudPolicy on exec-aide table
  • secretsmanager:GetSecretValue on exec-aide/gmail-oauth, exec-aide/slack-bot-token
  • secretsmanager:PutSecretValue on exec-aide/gmail-oauth (token refresh writes back)
  • ssm:GetParameter on /exec-aide/*
  • bedrock:InvokeModel on anthropic.* foundation models

daily-digest:

  • DynamoDBReadPolicy on exec-aide table (+ write for digest-sent marker)
  • secretsmanager:GetSecretValue on exec-aide/gmail-oauth, exec-aide/slack-bot-token
  • secretsmanager:PutSecretValue on exec-aide/gmail-oauth
  • ssm:GetParameter on /exec-aide/*

CloudFormation Outputs

  • FetchClassifyFunctionArn
  • DailyDigestFunctionArn
  • ExecAideTableName

Secrets & Config

Secrets Manager (credentials)

Secret Contents
exec-aide/gmail-oauth { client_id, client_secret, refresh_token, access_token, token_expiry }
exec-aide/slack-bot-token Bot token string (xoxb-...)

SSM Parameter Store (non-secret config)

Parameter Type Value
/exec-aide/adam-email String adam@seahavenind.com
/exec-aide/adam-slack-user-id String U01XXXXXXXX
/exec-aide/vip-senders String JSON array of email addresses
/exec-aide/vip-domains String JSON array of domains
/exec-aide/work-order-addresses String JSON array: ["work-orders@seahavenind.com","work-orders@seahaven.com"]
/exec-aide/unanswered-threshold-hours String 24

Module Design

shared/secrets.py

Module-level cached globals. get_gmail_oauth(), get_slack_token(), get_config(). Writes updated Gmail tokens back via save_gmail_tokens().

shared/gmail.py

  • get_authenticated_service(): Refreshes access token via google.oauth2.credentials.Credentials, persists new token to Secrets Manager, returns Gmail API service object.
  • fetch_history(history_id): Calls users.history.list with historyTypes=messageAdded, labelIds=INBOX. Returns (message_ids, new_history_id).
  • fetch_message(message_id): Calls users.messages.get with format=metadata. Parses headers into structured dict.
  • fetch_thread(thread_id): Gets all messages in a thread — used by daily digest to verify unanswered status.
  • initial_sync(): First-run fallback. users.messages.list with q=newer_than:2d&labelIds=INBOX.

shared/classify.py

  • classify_email(message, vip_senders, vip_domains): Calls Bedrock Converse API with Haiku. Returns { classification, reason }.
  • check_bypassed_workorder(message, work_order_addresses): Pure logic — checks if TO/CC contains work-orders addresses. No AI needed.
  • Prompt returns JSON: { "classification": "HIGH"|"NORMAL"|"LOW", "reason": "..." }

shared/dynamo.py

  • save_message(): PutItem with ConditionExpression: attribute_not_exists(pk) for idempotency.
  • save_thread_state(): UpdateItem on THD# — tracks last message sender/date and Adam's last reply.
  • message_exists(): Dedup check before processing.
  • get_todays_messages(): GSI query by classified_date.
  • get_unanswered_threads(): Query THD# items where unanswered_since is set.
  • get/save_sync_metadata(): Read/write META#sync.

shared/slack.py

  • send_dm(blocks, text): Opens DM via conversations.open, posts via chat.postMessage. Raw HTTP with requests (matches existing patterns — no slack_sdk).
  • build_high_priority_alert(message): Block Kit for real-time alert — header, sender/time fields, subject+snippet, "Open in Gmail" button.
  • build_daily_digest(high_items, bypassed, unanswered, stats): Block Kit with sections for each category, stats footer.

Handler Logic

fetch_classify/app.py (every 15 min)

  1. Load secrets + config (cached on warm start)
  2. Read META#sync from DynamoDB
  3. If no sync metadata → initial_sync(); else → fetch_history(history_id)
  4. For each new message:
    • Skip if message_exists() (dedup)
    • fetch_message() for metadata
    • classify_email() via Bedrock
    • check_bypassed_workorder() (pure logic)
    • save_message() to DynamoDB
    • save_thread_state() to DynamoDB
    • If HIGH → send_dm() immediately
  5. Update META#sync with new history_id
  6. Return summary

Error resilience: Each message processed independently. Partial failures logged, don't block remaining messages. History ID only updated after successful processing. If history.list returns 404 (expired ID), falls back to initial_sync().

daily_digest/app.py (5 PM ET, weekdays)

  1. Load secrets + config
  2. Query today's messages from GSI by-date
  3. Query THD# items with unanswered_since set
  4. For unanswered threads: call fetch_thread() via Gmail API to verify Adam hasn't replied (catches replies sent via phone/web that the 15-min poll missed)
  5. Filter bypassed work-orders from today's messages
  6. Compute stats (total, by classification)
  7. Build digest Block Kit message
  8. Send DM to Adam
  9. Log digest-sent marker to DynamoDB

Gmail Incremental Sync

  • Initial sync: messages.list with q=newer_than:2d&labelIds=INBOX. Fetch each message. Store latest historyId.
  • Incremental: history.list with startHistoryId. Process only messagesAdded. Store new historyId.
  • History expired (404): Fall back to initial sync. Idempotency prevents double-processing.

Unanswered Thread Detection

  • During fetch_classify: When a new message arrives, update THD#<threadId>. If from Adam → clear unanswered_since. If from someone else and Adam hasn't replied after this message → set unanswered_since.
  • During daily_digest: Re-verify via Gmail threads.get to catch replies the poll missed (sent from phone, web). Only include truly unanswered threads older than threshold.

One-Time Setup Steps

Gmail OAuth

  1. Google Cloud Console → create/reuse project → enable Gmail API
  2. OAuth consent screen (Internal for Workspace)
  3. Create OAuth Client ID (Web app), redirect URI http://localhost:8080/callback
  4. Run scripts/get_gmail_token.py → authorize as adam@seahavenind.com
  5. Store credentials in Secrets Manager: exec-aide/gmail-oauth
  6. Scope: gmail.readonly

Slack App

  1. api.slack.com/apps → Create New App → "Exec Aide" in Sea Haven workspace
  2. Scopes: chat:write, im:write
  3. Install to workspace
  4. Store bot token in Secrets Manager: exec-aide/slack-bot-token
  5. Store Adam's user ID in SSM: /exec-aide/adam-slack-user-id
  6. Update Notion Slack Apps Inventory page

SSM Parameters

Create all /exec-aide/* parameters listed above.

Implementation Order

  1. Scaffold repo, template.yaml, .gitignore, samconfig.toml.example
  2. Gmail OAuth setup (one-time, needed before testing)
  3. Slack app setup (one-time)
  4. Secrets Manager + SSM parameter creation
  5. shared/secrets.py (foundational)
  6. shared/gmail.py + shared/dynamo.py
  7. shared/classify.py
  8. shared/slack.py
  9. fetch_classify/app.py
  10. daily_digest/app.py
  11. sam build && sam deploy
  12. Smoke test: manual invoke → check DynamoDB → verify Slack DM
  13. README
  14. Notion: AWS Architecture Map + Slack Apps Inventory updates
  15. Project memory entry

Verification

  1. sam build succeeds
  2. sam deploy — stack creates without errors
  3. Manual invoke of fetch-classify → check CloudWatch logs for successful Gmail sync → DynamoDB has MSG and META items
  4. Send test email with "URGENT" subject → wait 15 min (or manual invoke) → Slack DM received
  5. Manual invoke of daily-digest → Slack digest DM received with correct sections
  6. Send email directly to Adam without CC'ing work-orders@ → verify it appears in "Bypassed Work Orders" section of next digest
  7. Leave an email unanswered for 24h → verify it appears in "Unanswered Threads" section