feat: implement Redis and Valkey schedule stores, add SREM method to …

.agents/skills/ak-dev-testing-conventions/SKILL.md

@@ -100,8 +100,9 @@ Tests live in `ak-py/tests/` and follow the naming convention `test_<module>.py`
 | `test_authoriser_shared.py` | Shared `Authoriser` in `agentkernel.auth`: `AuthValidatorAuthoriser` adaptation, export identity, and the guard that the thread package no longer re-exports it |
 | `test_schedule_model.py` | `ScheduleSpec` one-of/timezone/session_mode validation, chat-envelope parsing, `ScheduledTask` JSON round trip (JSON primitives only) |
 | `test_schedule_manager.py` | `ScheduleManager`: `get()` gating + singleton, provider/transport fail-fast, semantic validation matrix (including the named-agent precheck and the unnamed-agent exemption), create ordering + rollback, trigger-body freezing, ownership, amendment rules (occurrence rule replaced as a unit, untouched when the amendment names none of it), cancellation, occurrence recording (never raises) |
-| `test_schedule_store.py` | `ScheduleStore` backends (in_memory today) + `ScheduleStoreBuilder` built-in/BYO/unknown-type resolution |
+| `test_schedule_store.py` | Every `ScheduleStore` backend against the one contract the in_memory class pins — in_memory, redis/valkey through a fake redis-like client injected as `store._driver._client`, dynamodb through a fake driver whose `table.scan` replays `LastEvaluatedKey` pages — plus index cleanup on delete, TTL behaviour (none by default), and `ScheduleStoreBuilder` built-in/BYO/unknown-type/missing-extra resolution |
 | `test_schedule_provider_local.py` | `LocalScheduleProvider`: next-fire computation, one-time vs re-armed occurrences, token substitution, body-only delivery into `InMemoryTransport`, pause/delete disarm, `ScheduleProviderFactory` resolution |
+| `test_schedule_provider_eventbridge.py` | `EventBridgeScheduleProvider` against a fake `scheduler` client (monkeypatched at the module's own `boto3.client`): exact create/update/delete/get kwargs, `at()` and 5→6-field cron translation with each `?` day case, `ActionAfterCompletion`/`State`/timezone mapping, context-attribute substitution in the target `Input`, FIFO group per session mode, ARN-derived group/name, `ClientError` → `ScheduleError` and the tolerated not-found, plus the factory branch and its incomplete-config error |
 | `test_schedule_router.py` | `ScheduleRESTRequestHandler`: 404 when unconfigured, the three 401 variants from the shared `AuthorisedRESTRequestHandler`, listings forced to the authorised user, 403-before-404 ordering, PUT amendment happy path + validation 400s, DELETE returning the cancelled task, and the guard that importing the package does not pull in FastAPI |
 | `test_schedule_tools.py` | Schedule system tools: `SystemToolFactory` registration + `schedule.agents` scoping, prompt-suffix content, disabled short-circuit, acting-user read from the session volatile cache, and each tool's JSON contract including the no-identity and unknown-agent errors |
 | `test_chat_service_schedule.py` | ChatService interception on all four entry points: 202 wire shapes, streaming terminal chunk, unconfigured 400, occurrence recording, no scheduling field leaking as `AgentRequestAny` |

.github/integration-test-config.yaml

@@ -52,6 +52,12 @@ weekly:
     - type: aws-containerized
       path: examples/aws-containerized/openai-dynamodb-scalable
       deploy_dir: deploy
+    # Scheduled/recurring chats: EventBridge Scheduler + DynamoDB schedule store.
+    # The test suite cancels every schedule it creates (and any the agent registered) on teardown —
+    # terraform destroy cannot delete a schedule group that still holds schedules.
+    - type: aws-containerized
+      path: examples/aws-containerized/openai-schedule
+      deploy_dir: deploy
 
     # AWS Serverless
     - type: aws-serverless
@@ -69,6 +75,11 @@ weekly:
     - type: aws-serverless
       path: examples/aws-serverless/openai-auth
       deploy_dir: deploy
+    # Scheduled/recurring chats, all three Lambdas in zip mode. Management routes are the custom
+    # Lambda.register routes (the serverless router is not FastAPI), so the suite drives those.
+    - type: aws-serverless
+      path: examples/aws-serverless/schedule-openai
+      deploy_dir: deploy
 
     # Memory options
     - type: aws-serverless

README.md

@@ -124,12 +124,24 @@ Let agents run code and shell commands in an isolated, permission-bounded enviro
 
 [Learn more →](https://kernel.yaala.ai/docs/advanced/sandbox)
 
+### ⏰ Deferred & Recurring Chats
+
+Let a chat run later, or on a schedule — the platform owns the timers, the persistence, and the management API.
+
+- **Enable it in config** — a `schedule` block is the whole switch; no handler to mount, no code change.
+- **One creation path, three callers** — a `schedule` block on any chat request (acknowledged with HTTP 202), the agent's own `create_schedule` tool, or a direct `ScheduleManager` call.
+- **Pluggable timers and stores** — `local` (in-process, for development) or AWS EventBridge Scheduler for production; task records in memory, Redis, Valkey, or DynamoDB.
+- **Managed over REST** — list, read, amend, pause and cancel via `/api/v1/schedules`, scoped to the owning user by a pluggable `Authoriser`.
+
+[Learn more →](https://kernel.yaala.ai/docs/advanced/scheduling)
+
 ### 🧠 Memory, Sessions & Knowledge Bases
 
 | Layer | Backends |
 |---|---|
 | **Session / Memory** | In-memory, Redis, Valkey (AWS), DynamoDB (AWS), Cosmos DB (Azure), Firestore (GCP) |
 | **Conversation Threads** | Persistent, named threads keyed by `session_id` — in-memory, Redis, Valkey, DynamoDB (AWS), Cosmos DB (Azure), Firestore (GCP) |
+| **Scheduled Tasks** | Deferred and recurring chat execution — in-memory, Redis, Valkey, DynamoDB (AWS) task stores; local in-process or AWS EventBridge Scheduler timers |
 | **Vector Knowledge** | ChromaDB |
 | **Graph Knowledge** | Neo4j |
 | **SQL Analytics** | Starburst Galaxy (Trino) |
@@ -179,7 +191,7 @@ ak skill install
 |---|---|
 | `ak-init` | Scaffold a new project — any framework, any deployment mode |
 | `ak-build` | Add tools, agents, handoffs — context-aware and framework-specific |
-| `ak-add-capabilities` | Wire in guardrails, tracing, sessions, MCP, A2A, hooks, multimodal, conversation threads |
+| `ak-add-capabilities` | Wire in guardrails, tracing, sessions, MCP, A2A, hooks, multimodal, conversation threads, scheduled tasks |
 | `ak-add-integration` | Slack, WhatsApp, Messenger, Instagram, Telegram, Gmail |
 | `ak-cloud-deploy` | AWS Lambda, ECS, Azure Functions, Container Apps, GCP Cloud Run with full Terraform |
 | `ak-test` | Fuzzy, judge, and fallback test modes + a debugging playbook |

ak-deployment/ak-aws/containerized/README.md

@@ -316,6 +316,46 @@ enable_api_gateway_logs = true
 
   Skipping this makes Terraform destroy and recreate these resources, discarding any retained logs.
 
+### Scheduling (EventBridge Scheduler)
+
+| Variable | Description | Type | Default | Required |
+|---|---|---|---|---|
+| `enable_scheduling` | Create the EventBridge Scheduler schedule group and the execution role Scheduler assumes to deliver triggers to the Input Queue, grant both ECS task roles `scheduler:*Schedule` + `iam:PassRole` on them, and inject their coordinates. **Requires `queue_mode = true`.** | `bool` | `false` | no |
+| `create_dynamodb_schedule_table` | Create the DynamoDB schedule store table (partition `task_id`, no sort key, no GSI, TTL on `expiry_time`) and inject its generated name as `AK_SCHEDULE__STORE__DYNAMODB__TABLE_NAME` | `bool` | `false` | no |
+
+```hcl
+queue_mode                     = true
+enable_scheduling              = true
+create_dynamodb_schedule_table = true
+```
+
+- Off by default; every resource is `count`-gated, so leaving both `false` provisions nothing and
+  injects nothing.
+- **You must also declare the backends in the application's `config.yaml`** — Terraform injects the
+  group/role/queue/table coordinates but never `schedule.provider.type` or `schedule.store.type`
+  (the same rule as `thread.type`):
+
+  ```yaml
+  schedule:
+    provider:
+      type: eventbridge
+    store:
+      type: dynamodb
+  ```
+
+  Setting the flags without this block leaves scheduling on the default `local` provider and
+  `in_memory` store, and the provisioned group and table sit unused with no error.
+- `enable_scheduling` flips the **Input Queue** to `content_based_deduplication = true` (an in-place
+  update on an existing queue). EventBridge Scheduler cannot set a `MessageDeduplicationId`, so
+  without it two occurrences carrying an otherwise identical trigger body would collapse into one
+  inside the 5-minute dedup window. Application senders are unaffected: they always send an explicit
+  `MessageDeduplicationId`, which takes precedence. The Output Queue is untouched.
+- Both task roles get the schedule permissions: the REST service serves the management routes
+  (amend/cancel reach Scheduler), and the agent runner hosts the `create_schedule` /
+  `update_schedule` / `delete_schedule` agent tools.
+- See the [scheduling guide](https://kernel.yaala.ai/docs/advanced/scheduling) for the application
+  side.
+
 ## Deployment Modes
 
 ### Non-Queue Mode (Default)
@@ -587,6 +627,13 @@ output "websocket_api_stage_name"         # WebSocket API Gateway stage name
 output "websocket_endpoint_url"           # Management API endpoint used for PostToConnection
 output "websocket_connection_table_name"  # DynamoDB connections table name
 output "websocket_connection_table_arn"   # DynamoDB connections table ARN
+
+# Scheduling only (`enable_scheduling` / `create_dynamodb_schedule_table`) — null otherwise
+output "schedule_group_name"           # EventBridge Scheduler schedule-group name
+output "schedule_group_arn"            # EventBridge Scheduler schedule-group ARN
+output "scheduler_execution_role_arn"  # Role Scheduler assumes to deliver triggers to the Input Queue
+output "schedule_table_name"           # DynamoDB schedule store table name
+output "schedule_table_arn"            # DynamoDB schedule store table ARN
 ```
 
 ## Requirements

ak-deployment/ak-aws/containerized/dynamodb.tf

@@ -21,6 +21,32 @@ resource "aws_dynamodb_table" "response_store" {
   tags = merge(var.tags, { Type = "ResponseStore" })
 }
 
+# DynamoDB Schedule Store
+# Persists the scheduled task records. One item per task, no sort key and no GSI: listings scan with
+# a filter expression, which is acceptable at schedule cardinalities.
+
+resource "aws_dynamodb_table" "schedule_store" {
+  count = var.create_dynamodb_schedule_table ? 1 : 0
+
+  name         = "${local.prefix}-schedule-store"
+  billing_mode = "PAY_PER_REQUEST"
+  hash_key     = "task_id"
+
+  attribute {
+    name = "task_id"
+    type = "S"
+  }
+
+  # Enabled so a deployment that opts into `schedule.store.dynamodb.ttl` works; the application's
+  # TTL defaults to 0, in which case items carry no `expiry_time` and never expire.
+  ttl {
+    attribute_name = "expiry_time"
+    enabled        = true
+  }
+
+  tags = merge(var.tags, { Type = "ScheduleStore" })
+}
+
 # WebSocket connections table, maps user_id <-> connection_id (WebSocket modes only)
 
 module "websocket_connections" {

ak-deployment/ak-aws/containerized/eventbridge.tf

@@ -0,0 +1,63 @@
+# EventBridge Scheduler resources for the scheduling capability.
+# The application owns the schedules themselves (one per scheduled task, created at runtime through
+# `AKConfig.schedule.provider.type: eventbridge`); Terraform only provisions the group they live in
+# and the role Scheduler assumes to deliver each trigger to the Input Queue.
+
+check "scheduling_requires_queue_mode" {
+  assert {
+    condition     = var.enable_scheduling ? var.queue_mode : true
+    error_message = "[IMPORTANT] enable_scheduling requires queue_mode = true: EventBridge Scheduler delivers its triggers to the Input Queue, which only exists in queue mode."
+  }
+}
+
+resource "aws_scheduler_schedule_group" "schedules" {
+  count = var.enable_scheduling ? 1 : 0
+
+  name = "${local.prefix}-schedules"
+
+  tags = merge(var.tags, { Type = "ScheduleGroup" })
+}
+
+# Execution role Scheduler assumes per occurrence to send the trigger to the Input Queue.
+# The application passes this role's ARN when it registers a schedule, which is why both task roles
+# also need iam:PassRole on it (see iam.tf and modules/agent-runner/main.tf).
+
+resource "aws_iam_role" "scheduler_execution" {
+  count = var.enable_scheduling ? 1 : 0
+
+  name        = "${local.prefix}-scheduler-exec-role"
+  description = "Role EventBridge Scheduler assumes to deliver scheduled triggers to the Input Queue"
+
+  assume_role_policy = jsonencode({
+    Version = "2012-10-17"
+    Statement = [{
+      Effect    = "Allow"
+      Action    = "sts:AssumeRole"
+      Principal = { Service = "scheduler.amazonaws.com" }
+      Condition = {
+        StringEquals = {
+          "aws:SourceAccount" = data.aws_caller_identity.current.account_id
+        }
+      }
+    }]
+  })
+
+  tags = var.tags
+}
+
+resource "aws_iam_role_policy" "scheduler_send_to_input_queue" {
+  count = var.enable_scheduling ? 1 : 0
+
+  name = "${local.prefix}-scheduler-send-to-input-queue"
+  role = aws_iam_role.scheduler_execution[0].id
+
+  # No KMS statement: the queues use SQS-managed SSE, not a customer-managed key.
+  policy = jsonencode({
+    Version = "2012-10-17"
+    Statement = [{
+      Effect   = "Allow"
+      Action   = ["sqs:SendMessage"]
+      Resource = module.queues[0].input_queue_arn
+    }]
+  })
+}

ak-deployment/ak-aws/containerized/iam.tf

@@ -68,6 +68,70 @@ resource "aws_iam_policy" "rest_service_response_store_policy" {
   tags = var.tags
 }
 
+# Scheduling IAM policies (management routes: amend/cancel reach EventBridge Scheduler)
+
+resource "aws_iam_policy" "rest_service_scheduler_policy" {
+  count = var.enable_scheduling ? 1 : 0
+
+  name        = "${local.prefix}-rest-svc-scheduler"
+  description = "Allow REST Service ECS task to manage EventBridge schedules in the AK schedule group"
+
+  policy = jsonencode({
+    Version = "2012-10-17"
+    Statement = [
+      {
+        Sid    = "ManageSchedules"
+        Effect = "Allow"
+        Action = [
+          "scheduler:CreateSchedule",
+          "scheduler:UpdateSchedule",
+          "scheduler:DeleteSchedule",
+          "scheduler:GetSchedule"
+        ]
+        Resource = "arn:aws:scheduler:*:${data.aws_caller_identity.current.account_id}:schedule/${local.schedule_group_name}/*"
+      },
+      {
+        # Scheduler assumes the execution role, so registering a schedule passes it.
+        Sid      = "PassSchedulerExecutionRole"
+        Effect   = "Allow"
+        Action   = ["iam:PassRole"]
+        Resource = local.scheduler_execution_role_arn
+      }
+    ]
+  })
+
+  tags = var.tags
+}
+
+resource "aws_iam_policy" "rest_service_schedule_store_policy" {
+  count = var.create_dynamodb_schedule_table ? 1 : 0
+
+  name        = "${local.prefix}-rest-svc-schedule-store"
+  description = "Allow REST Service ECS task to read/write the DynamoDB schedule store"
+
+  policy = jsonencode({
+    Version = "2012-10-17"
+    Statement = [
+      {
+        Effect = "Allow"
+        Action = [
+          "dynamodb:DescribeTable",
+          "dynamodb:GetItem",
+          "dynamodb:PutItem",
+          "dynamodb:UpdateItem",
+          "dynamodb:DeleteItem",
+          "dynamodb:Query",
+          "dynamodb:Scan"
+        ]
+        # No /index/* : listings Scan, this table has no GSI.
+        Resource = local.dynamodb_schedule_table_arn
+      }
+    ]
+  })
+
+  tags = var.tags
+}
+
 resource "aws_iam_role_policy_attachment" "rest_service_sqs_attachment" {
   count      = var.queue_mode ? 1 : 0
   role       = module.rest_service.task_role_name
@@ -80,6 +144,18 @@ resource "aws_iam_role_policy_attachment" "rest_service_response_store_attachmen
   policy_arn = aws_iam_policy.rest_service_response_store_policy[0].arn
 }
 
+resource "aws_iam_role_policy_attachment" "rest_service_scheduler_attachment" {
+  count      = var.enable_scheduling ? 1 : 0
+  role       = module.rest_service.task_role_name
+  policy_arn = aws_iam_policy.rest_service_scheduler_policy[0].arn
+}
+
+resource "aws_iam_role_policy_attachment" "rest_service_schedule_store_attachment" {
+  count      = var.create_dynamodb_schedule_table ? 1 : 0
+  role       = module.rest_service.task_role_name
+  policy_arn = aws_iam_policy.rest_service_schedule_store_policy[0].arn
+}
+
 # REST service WebSocket IAM policies (async / stream modes)
 
 # Push messages to connected clients (PostToConnection).

ak-deployment/ak-aws/containerized/modules/README.md

@@ -152,6 +152,15 @@ rest_service = {
 - `AK_EXECUTION__QUEUES__BATCH_SIZE` (if queue mode enabled, from root `queue_config.batch_size`)
 - `AK_EXECUTION__MODE` (`async` or `stream`, WebSocket mode only)
 - `AK_WEBSOCKET_API__CHAT_ROUTE`, `AK_WEBSOCKET_API__ENDPOINT_URL`, `AK_WEBSOCKET_API__CONNECTION_TABLE__TABLE_NAME` (WebSocket mode only)
+- `AK_SCHEDULE__PROVIDER__EVENTBRIDGE__GROUP_NAME`, `__ROLE_ARN`, `__QUEUE_ARN` (if `enable_scheduling`)
+- `AK_SCHEDULE__STORE__DYNAMODB__TABLE_NAME` (if `create_dynamodb_schedule_table`)
+
+  > Terraform never injects `AK_SCHEDULE__PROVIDER__TYPE` or `AK_SCHEDULE__STORE__TYPE` — the
+  > application declares `schedule.provider.type: eventbridge` and `schedule.store.type: dynamodb`
+  > in its committed `config.yaml`, the same rule as `thread.type`. Because any `AK_SCHEDULE__*`
+  > variable is enough to populate the config block, the injected variables *alone* would enable
+  > scheduling on the default `local`/`in_memory` backends: the provisioned group and table would
+  > sit unused.
 
 #### WebSocket Mode
 
@@ -238,6 +247,8 @@ scaling_config = {
 - `AK_SESSION__DYNAMODB__TABLE_NAME` (if DynamoDB memory table enabled)
 - `AK_EXECUTION__MODE` (`async` or `stream`, WebSocket mode only — lets the agent runner forward
   the `endpoint_url` custom attribute to the Output Queue so the REST/IO service can push the reply)
+- `AK_SCHEDULE__PROVIDER__EVENTBRIDGE__GROUP_NAME`, `__ROLE_ARN`, `__QUEUE_ARN` (if `enable_scheduling`)
+- `AK_SCHEDULE__STORE__DYNAMODB__TABLE_NAME` (if `create_dynamodb_schedule_table`)
 
 **Auto Scaling Behavior**:
 When `scaling_config.enabled = true`: