Add tasks and reminders

- DynamoDB task CRUD with TASK#{ulid} prefix
- 5 new tools: create_task, list_tasks, complete_task, delete_task, create_reminder
- New exec-aide-reminder Lambda triggered by EventBridge Scheduler one-shots
- CDK: reminder Lambda, scheduler IAM role, conversation Lambda permissions
- Updated system prompt with task/reminder capabilities

Closes #3

lib/constructs/email-pipeline.ts

@@ -150,6 +150,31 @@ export class EmailPipelineConstruct extends Construct {
       sourceArn: `arn:aws:scheduler:${region}:${account}:schedule/default/${schedule.name}`,
     });
 
+    // ── Reminder Lambda ──────────────────────────────────────
+
+    const reminder = new PythonFunction(this, 'Reminder', {
+      functionName: 'exec-aide-reminder',
+      entry: path.join(__dirname, '../../src'),
+      index: 'reminder/app.py',
+      handler: 'lambda_handler',
+      runtime: lambda.Runtime.PYTHON_3_12,
+      architecture: lambda.Architecture.ARM_64,
+      memorySize: 128,
+      timeout: cdk.Duration.seconds(30),
+      environment: lambdaEnv,
+      logRetention: logs.RetentionDays.TWO_MONTHS,
+    });
+
+    this.table.grantReadData(reminder);
+    reminder.addToRolePolicy(secretsReadPolicy);
+    reminder.addToRolePolicy(ssmPolicy);
+
+    const reminderSchedulerRole = new iam.Role(this, 'ReminderSchedulerRole', {
+      roleName: 'exec-aide-reminder-scheduler',
+      assumedBy: new iam.ServicePrincipal('scheduler.amazonaws.com'),
+    });
+    reminder.grantInvoke(reminderSchedulerRole);
+
     // ── Conversation Lambda ───────────────────────────────────
 
     const conversation = new PythonFunction(this, 'Conversation', {
@@ -161,7 +186,11 @@ export class EmailPipelineConstruct extends Construct {
       architecture: lambda.Architecture.ARM_64,
       memorySize: 512,
       timeout: cdk.Duration.seconds(180),
-      environment: lambdaEnv,
+      environment: {
+        ...lambdaEnv,
+        REMINDER_FN_ARN: reminder.functionArn,
+        REMINDER_SCHEDULER_ROLE_ARN: reminderSchedulerRole.roleArn,
+      },
       logRetention: logs.RetentionDays.TWO_MONTHS,
     });
 
@@ -176,6 +205,16 @@ export class EmailPipelineConstruct extends Construct {
         `arn:aws:bedrock:${region}:${account}:inference-profile/us.anthropic.*`,
       ],
     }));
+    conversation.addToRolePolicy(new iam.PolicyStatement({
+      actions: ['scheduler:CreateSchedule', 'scheduler:DeleteSchedule'],
+      resources: [
+        `arn:aws:scheduler:${region}:${account}:schedule/default/exec-aide-reminder-*`,
+      ],
+    }));
+    conversation.addToRolePolicy(new iam.PolicyStatement({
+      actions: ['iam:PassRole'],
+      resources: [reminderSchedulerRole.roleArn],
+    }));
     dailyDigest.grantInvoke(conversation);
 
     this.conversationFn = conversation;

src/conversation/app.py

@@ -37,8 +37,16 @@ unnecessary preamble. Use bullet points for lists of emails.
 When referencing emails, always include the sender name and subject line. \
 For time references, use relative terms (e.g., "2 hours ago", "yesterday").
 
-If Adam asks about something outside your current capabilities (calendar, \
+You can manage Adam's task list — create tasks, list them, mark them \
-tasks, etc.), acknowledge it and let him know it's not available yet.
+complete, or delete them. You can also set reminders that will send a Slack \
+DM at a specified time. When Adam asks to be reminded of something, use \
+create_reminder with the appropriate date/time (default timezone: \
+America/New_York).
+
+You have access to Adam's Google Calendar. You can view upcoming events, \
+check availability, and create new events. Always confirm event details \
+(time, attendees, title) with Adam before creating. Default timezone: \
+America/New_York. Use ISO 8601 format for all times sent to calendar tools.
 
 Speak naturally and casually, like a capable coworker - not a bot or AI. \
 Keep responses conversational.

src/reminder/app.py

@@ -0,0 +1,25 @@
+import logging
+
+from shared.secrets import get_config, get_slack_token
+from shared.slack import post_message
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+
+def lambda_handler(event, context):
+    text = event.get("text", "Reminder (no details provided)")
+    config = get_config()
+    token = get_slack_token()
+
+    import requests
+    resp = requests.post(
+        "https://slack.com/api/conversations.open",
+        headers={"Authorization": f"Bearer {token}"},
+        json={"users": config["adam_slack_user_id"]},
+    )
+    resp.raise_for_status()
+    channel = resp.json()["channel"]["id"]
+
+    post_message(channel, f":bell: *Reminder:* {text}", token)
+    logger.info("Reminder sent: %s", text)

src/requirements.txt

@@ -1,3 +1,4 @@
 google-api-python-client>=2.195.0
 google-auth>=2.50.0
+python-ulid>=2.0
 requests>=2.33.1

src/shared/dynamo.py

@@ -3,6 +3,7 @@ import time
 from datetime import datetime, timezone
 
 import boto3
+import ulid
 from botocore.exceptions import ClientError
 
 _table = None
@@ -205,3 +206,62 @@ def get_unanswered_threads(threshold_hours):
         item for item in items
         if datetime.fromisoformat(item["unanswered_since"]).timestamp() < cutoff
     ]
+
+
+# ── Tasks ──────────────────────────────────────────────────────
+
+
+def create_task(title, due_date=None):
+    task_id = str(ulid.new())
+    _get_table().put_item(Item={
+        "pk": f"TASK#{task_id}",
+        "sk": "TASK",
+        "title": title,
+        "due_date": due_date or "",
+        "status": "open",
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "completed_at": "",
+        "ttl": _ttl_90_days(),
+    })
+    return task_id
+
+
+def get_tasks(status_filter=None):
+    scan_kwargs = {
+        "FilterExpression": "begins_with(pk, :prefix)",
+        "ExpressionAttributeValues": {":prefix": "TASK#"},
+    }
+    if status_filter:
+        scan_kwargs["FilterExpression"] += " AND #s = :status"
+        scan_kwargs["ExpressionAttributeNames"] = {"#s": "status"}
+        scan_kwargs["ExpressionAttributeValues"][":status"] = status_filter
+
+    items = []
+    resp = _get_table().scan(**scan_kwargs)
+    items.extend(resp.get("Items", []))
+    while "LastEvaluatedKey" in resp:
+        scan_kwargs["ExclusiveStartKey"] = resp["LastEvaluatedKey"]
+        resp = _get_table().scan(**scan_kwargs)
+        items.extend(resp.get("Items", []))
+
+    items.sort(key=lambda x: x["pk"])
+    return items
+
+
+def update_task_status(task_id, status):
+    update_expr = "SET #s = :status, ttl = :ttl"
+    expr_values = {":status": status, ":ttl": _ttl_90_days()}
+    if status == "completed":
+        update_expr += ", completed_at = :now"
+        expr_values[":now"] = datetime.now(timezone.utc).isoformat()
+
+    _get_table().update_item(
+        Key={"pk": f"TASK#{task_id}", "sk": "TASK"},
+        UpdateExpression=update_expr,
+        ExpressionAttributeNames={"#s": "status"},
+        ExpressionAttributeValues=expr_values,
+    )
+
+
+def delete_task(task_id):
+    _get_table().delete_item(Key={"pk": f"TASK#{task_id}", "sk": "TASK"})