light weight scheduler and outbox sweeper

Makefile

@@ -57,7 +57,7 @@ dev-web:
 
 dev-worker-orchestrator:
 	@echo "Starting Orchestrator Worker for local development..."
-	ENVIRONMENT=development PYTHONPATH=services/workers/orchestrator/src:libs/database/src:libs/config/src:libs/pipeline/src:libs/domain/src:libs/transformer/src uv run python services/workers/orchestrator/src/worker/main.py
+	ENVIRONMENT=development PYTHONPATH=services/workers/orchestrator/src:libs/database/src:libs/config/src:libs/pipeline/src:libs/domain/src:libs/transformer/src:libs/scheduler/src uv run python services/workers/orchestrator/src/worker/main.py
 
 dev-worker-compute:
 	@echo "Starting Compute Worker for local development..."
@@ -92,7 +92,7 @@ db-sqs-reset: db-reset sqs-purge
 
 clear-data:
 	@echo "Clearing data plane tables (edi_message, edi_json, api_gateway, outbox) and purging SQS..."
-	uv run python scripts/clear_data.py
+	uv run python scripts/clear_data.py --i-am-sure
 
 seed: db-init
 

TECHNICAL_DEBT.md

@@ -22,16 +22,17 @@ To achieve the long-term vision of completely merging and modernizing `bots_core
 ## 2. Outbox Sweeper (CDC Fallback Relay)
 
 **Description:**
-The system currently relies exclusively on Debezium (CDC) reading the PostgreSQL Write-Ahead Log (WAL) to route `Outbox` events to SQS. If Debezium crashes, loses offsets, or experiences network partitioning, `PENDING` outbox events will be permanently trapped in the database, breaking the asynchronous event pipeline.
+The Outbox Sweeper has been implemented as a fallback to Debezium (CDC). It iterates over all shards to relay `PENDING` outbox events to SQS. However, the current implementation iterates over shards sequentially (`for shard in shards: await self._sweep_shard(...)`). As the number of shards grows in the multi-tenant architecture, this sequential sweep will take longer and potentially exceed the polling interval, causing lag.
 
 **Proposed Resolution:**
-Implement an Outbox Sweeper background worker that acts as a robust enterprise fallback and garbage collector:
-1. **Fallback Poller:** A cron/scheduled task that periodically queries `SELECT * FROM outbox WHERE status = 'PENDING'` for events older than a configured threshold (e.g., 60 seconds) and manually relays them to SQS.
-2. **Garbage Collector:** A cleanup task that runs `DELETE FROM outbox WHERE status = 'COMPLETED'` for events older than 7 days to prevent unbounded database growth.
+Refactor the Outbox Sweeper to use Bounded Concurrency or Distributed Job Fan-out:
+1. **Bounded Concurrency:** Run sweeps concurrently using `asyncio.gather` bounded by an `asyncio.Semaphore` so multiple shards are swept at once without exhausting resources.
+2. **Distributed Job Fan-out:** Instead of a single job, spawn a `ScheduledJob` for each shard dynamically, allowing multiple orchestrator pods to load balance the shard sweeping.
+3. **Garbage Collector (Pending):** A cleanup task that runs `DELETE FROM outbox WHERE status = 'COMPLETED'` for events older than 7 days to prevent unbounded database growth is still needed.
 
-**Estimated Effort:** Low
+**Estimated Effort:** Low-Medium
 **Estimated Time:** 1 to 2 days
-**Impact:** Essential for enterprise-grade high availability. Guarantees no messages are ever lost due to CDC infrastructure failures and keeps the database optimized over time.
+**Impact:** Prevents the sweeper from falling behind as the number of database shards scales, ensuring enterprise-grade multi-tenant reliability.
 
 ## 3. AS2 Protocol
 
@@ -80,3 +81,19 @@ Currently, the bots engine does not support a lightweight validation mode (e.g.,
 ### UnitOfWork Architecture (Control Plane vs Data Plane Naming)
 Currently, the `UnitOfWork` (and its underlying SQL Alchemy repositories) leak infrastructure/deployment boundaries ("Control Plane" and "Data Plane") into domain business logic. We have giant God-objects like `SqlAlchemyControlPlaneRepository` inheriting from 10+ distinct repositories, causing namespace collisions and violating SOLID principles (Single Responsibility Principle).
 **Future Action:** Refactor `UnitOfWork` to remove `control_plane` and `data_plane` concepts from class names and properties. Use Composition to expose distinct Bounded Contexts (e.g., `self.trading_partners`, `self.transactions`, `self.routes`) instead of lumping them into control/data plane buckets.
+
+## 8. Hybrid SQS Tenancy (Dynamic Queue Resolution)
+
+**Description:**
+Currently, both inbound and outbound events are routed to a static, shared SQS queue (e.g., `TransformOrchestrationQueue`). In a multi-tenant environment, a massive batch of outbound events from one tenant can block critical inbound processing for all other tenants (the "noisy neighbor" problem).
+
+**Proposed Resolution:**
+Implement a Hybrid SQS routing model that dynamically resolves the target queue based on the tenant's tier and the event direction:
+1. **Dynamic Queue Resolver:** The CDC Relay and Outbox Sweeper should read the `tenant_id` from the outbox event and lookup the tenant's tier (cached in memory).
+2. **Standard Tenants:** Route inbound events to `standard-inbound-queue` and outbound events to `standard-outbound-queue`.
+3. **Enterprise Tenants:** Route events to dedicated queues (e.g., `enterprise-{tenant_name}-inbound-queue`).
+4. **Dedicated Workers:** Deploy separate worker pods for standard inbound, standard outbound, and dedicated enterprise queues to provide strict compute isolation.
+
+**Estimated Effort:** Medium
+**Estimated Time:** 3 to 5 days
+**Impact:** Essential for enterprise-grade SaaS scaling. Guarantees compute isolation for enterprise customers and bulkheads heavy outbound processing from blocking high-priority inbound traffic.

docker/localstack/init-aws.sh

@@ -14,7 +14,11 @@ awslocal sqs create-queue --queue-name CDC-DLQ
 
 awslocal sqs create-queue --queue-name TransformOrchestrationQueue-DLQ
 TRANSFORM_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/TransformOrchestrationQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
-awslocal sqs create-queue --queue-name TransformOrchestrationQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$TRANSFORM_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
+    awslocal sqs create-queue --queue-name TransformOrchestrationQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$TRANSFORM_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
+
+    awslocal sqs create-queue --queue-name TransformComputeQueue-DLQ
+    COMPUTE_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/TransformComputeQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
+    awslocal sqs create-queue --queue-name TransformComputeQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$COMPUTE_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
 
 awslocal sqs create-queue --queue-name DeliverQueue-DLQ
 DELIVER_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/DeliverQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
@@ -25,4 +29,8 @@ awslocal sqs create-queue --queue-name ProvisioningQueue-DLQ
 PROVISIONING_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/ProvisioningQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
 awslocal sqs create-queue --queue-name ProvisioningQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$PROVISIONING_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
 
+awslocal sqs create-queue --queue-name edi-orchestrator-jobs-DLQ
+JOBS_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/edi-orchestrator-jobs-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
+awslocal sqs create-queue --queue-name edi-orchestrator-jobs --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$JOBS_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
+
 echo "LocalStack Initialization Complete."

frontend/web/package.json

@@ -32,6 +32,7 @@
     "date-fns": "^4.4.0",
     "lucide-react": "^1.21.0",
     "next-themes": "^0.4.6",
+    "node-forge": "^1.4.0",
     "oidc-client-ts": "^3.5.0",
     "react": "^19.2.7",
     "react-dom": "^19.2.7",
@@ -44,6 +45,7 @@
   "devDependencies": {
     "@tanstack/router-plugin": "^1.168.19",
     "@types/node": "^24.13.2",
+    "@types/node-forge": "^1.3.14",
     "@types/react": "^19.2.17",
     "@types/react-dom": "^19.2.3",
     "@vitejs/plugin-react": "^6.0.2",

frontend/web/src/features/partners/api/IPartnersRepository.ts

@@ -17,6 +17,7 @@ import type {
  */
 export interface IPartnersRepository {
   // Platform Trading Partners
+  deleteCertificateSecret(vaultRef: string): Promise<void>;
   getPlatformPartners(): Promise<Partner[]>;
   createPlatformPartner(payload: CreatePartnerPayload): Promise<Partner>;
   updatePlatformPartner(id: string, payload: UpdatePartnerPayload): Promise<Partner>;
@@ -32,6 +33,7 @@ export interface IPartnersRepository {
   // Certificates
   exportCertificates(partnerId: string): Promise<CertificatesExport>;
   rotateCertificates(partnerId: string, payload: RotateCertPayload): Promise<Partner>;
+  generateCertificate(as2Id: string): Promise<{ public_cert_pem: string; private_key_vault_ref: string }>;
 
   // Tenant Partners
   getTenantPartners(): Promise<Partner[]>;

frontend/web/src/features/partners/api/partnerHooks.ts

@@ -127,16 +127,22 @@ export function useUpdatePlatformPartnerMutation() {
   );
 }
 
-export function useDeletePlatformPartner() {
+export function useDeletePlatformPartnerMutation() {
   const repo = useRepository();
-
   return useToastMutation(
     (id: string) => repo.deletePlatformPartner(id),
     'Partner deleted.',
     [partnersKeys.platformPartners()]
   );
 }
 
+export function useDeleteCertificateSecretMutation() {
+  const repo = useRepository();
+  return useMutation({
+    mutationFn: (vaultRef: string) => repo.deleteCertificateSecret(vaultRef),
+  });
+}
+
 // ─────────────────────────────────────────────
 // Platform Partnership Mutations
 // ─────────────────────────────────────────────
@@ -228,6 +234,13 @@ export function useRotateCertificatesMutation() {
   );
 }
 
+export function useGenerateCertificateMutation() {
+  const repo = useRepository();
+  return useMutation({
+    mutationFn: (as2Id: string) => repo.generateCertificate(as2Id)
+  });
+}
+
 export function useTestSftpConnectionMutation() {
   const repo = useRepository();
   return useMutation({

frontend/web/src/features/partners/api/partnersApi.ts

@@ -54,6 +54,10 @@ class HttpPartnersRepository implements IPartnersRepository {
   }
 
   // ── Platform Trading Partners ──────────────
+  deleteCertificateSecret(vaultRef: string): Promise<void> {
+    return this.request(`/api/v1/platform/trading-partners/as2/certificates/secret?vault_ref=${encodeURIComponent(vaultRef)}`, { method: 'DELETE' });
+  }
+
   getPlatformPartners(): Promise<Partner[]> {
     return this.request('/api/v1/platform/trading-partners/as2/trading-partners');
   }
@@ -124,6 +128,13 @@ class HttpPartnersRepository implements IPartnersRepository {
     );
   }
 
+  generateCertificate(as2Id: string): Promise<{ public_cert_pem: string; private_key_vault_ref: string }> {
+    return this.request(
+      `/api/v1/platform/trading-partners/as2/certificates/generate`,
+      { method: 'POST', body: JSON.stringify({ as2_id: as2Id }) },
+    );
+  }
+
   // ── Tenant Partners ────────────────────────
   async getTenantPartners(): Promise<Partner[]> {
     const data = await this.request<any[]>('/api/v1/trading-partners');