schema

Author: v1r3n

README.md

@@ -464,8 +464,14 @@ export conductor.worker.all.domain=production
 export conductor.worker.all.poll_interval_millis=250
 export conductor.worker.all.thread_count=20
 
+# Task registration configuration
+export conductor.worker.all.register_task_def=true  # Auto-register task definitions
+export conductor.worker.all.overwrite_task_def=true  # Overwrite existing (default)
+export conductor.worker.all.strict_schema=false  # Lenient schema validation (default)
+
 # Worker-specific configuration (overrides global)
 export conductor.worker.greetings.thread_count=50
+export conductor.worker.validate_order.strict_schema=true  # Strict validation for this worker
 
 # Runtime control (pause/resume workers without code changes)
 export conductor.worker.all.paused=true  # Maintenance mode

WORKER_CONFIGURATION.md

@@ -26,14 +26,18 @@ The following properties can be configured via environment variables:
 | `worker_id` | string | Unique worker identifier | `worker-1` | ✅ Yes |
 | `thread_count` | int | Max concurrent executions (threads for sync, coroutines for async) | `10` | ✅ Yes |
 | `register_task_def` | bool | Auto-register task definition with JSON schemas on startup | `true` | ✅ Yes |
+| `overwrite_task_def` | bool | Overwrite existing task definitions when registering (default: true) | `false` | ✅ Yes |
+| `strict_schema` | bool | Enforce strict schema validation - additionalProperties=false (default: false) | `true` | ✅ Yes |
 | `poll_timeout` | int | Poll request timeout in milliseconds | `100` | ✅ Yes |
 | `lease_extend_enabled` | bool | ⚠️ **Not implemented** - reserved for future use | `false` | ✅ Yes |
 | `paused` | bool | Pause worker from polling/executing tasks | `true` | ❌ **Environment-only** |
 
 **Notes**:
 - The `paused` property is intentionally **not available** in the `@worker_task` decorator. It can only be controlled via environment variables, allowing operators to pause/resume workers at runtime without code changes or redeployment.
 - The `lease_extend_enabled` parameter is accepted but **not currently implemented**. For lease extension, use manual `TaskInProgress` returns (see below).
-- The `register_task_def` parameter automatically registers task definitions with JSON Schema (draft-07) generated from Python type hints. Does not overwrite existing definitions.
+- The `register_task_def` parameter automatically registers task definitions with JSON Schema (draft-07) generated from Python type hints.
+- The `overwrite_task_def` parameter controls whether to overwrite existing task definitions (default: true).
+- The `strict_schema` parameter controls JSON schema validation strictness (default: false for lenient validation).
 
 ### Understanding `thread_count`
 
@@ -97,6 +101,69 @@ def long_task(job_id: str) -> Union[dict, TaskInProgress]:
 
 **For detailed patterns**, see [Long-Running Tasks & Lease Extension](docs/design/WORKER_DESIGN.md#long-running-tasks--lease-extension).
 
+### Understanding `overwrite_task_def`
+
+Controls whether to overwrite existing task definitions when `register_task_def=True`:
+
+**Overwrite Mode (default, overwrite_task_def=true):**
+- Always calls `update_task_def()` to overwrite existing definitions
+- Ensures server always has latest configuration from code
+- **Use when:** Task configuration changes frequently, development environments
+
+**No-Overwrite Mode (overwrite_task_def=false):**
+- Checks if task exists before registering
+- Only creates new task if it doesn't exist
+- Preserves manual changes made on server
+- **Use when:** Tasks managed outside code, production with manual config
+
+```bash
+# Global: Never overwrite any task definitions
+export conductor.worker.all.overwrite_task_def=false
+
+# Specific: Allow overwrite for this worker only
+export conductor.worker.dynamic_task.overwrite_task_def=true
+```
+
+### Understanding `strict_schema`
+
+Controls JSON Schema validation strictness when `register_task_def=True`:
+
+**Lenient Mode (default, strict_schema=false):**
+- Sets `additionalProperties=true` in schemas
+- Allows extra fields beyond defined schema
+- **Use when:** Backward compatibility, flexible integrations, development
+
+**Strict Mode (strict_schema=true):**
+- Sets `additionalProperties=false` in schemas
+- Rejects inputs with extra fields
+- **Use when:** Strict contract enforcement, production validation
+
+```bash
+# Global: Strict validation for all workers
+export conductor.worker.all.strict_schema=true
+
+# Specific: Lenient for this worker (overrides global)
+export conductor.worker.flexible_task.strict_schema=false
+```
+
+**Example Schemas:**
+
+```json
+// strict_schema=false (default)
+{
+  "type": "object",
+  "properties": {"name": {"type": "string"}},
+  "additionalProperties": true  // ← Extra fields allowed
+}
+
+// strict_schema=true
+{
+  "type": "object",
+  "properties": {"name": {"type": "string"}},
+  "additionalProperties": false  // ← Extra fields rejected
+}
+```
+
 ## Environment Variable Format
 
 ### Global Configuration (All Workers)

docs/design/WORKER_DESIGN.md

@@ -410,6 +410,8 @@ async def __get_authentication_headers(self):
 | `poll_timeout` | int | 100 | Poll timeout (ms) |
 | `lease_extend_enabled` | bool | False | ⚠️ **Not implemented** - use `TaskInProgress` instead |
 | `register_task_def` | bool | False | Auto-register task definition with JSON schemas (draft-07) |
+| `overwrite_task_def` | bool | True | Overwrite existing task definitions (when register_task_def=True) |
+| `strict_schema` | bool | False | Enforce strict schema validation (additionalProperties=false) |
 | `paused` | bool | False | Pause worker (env-only, not in decorator) |
 
 ### Examples
@@ -514,6 +516,80 @@ export conductor.worker.all.register_task_def=true
 
 # Enable for specific worker
 export conductor.worker.process_order.register_task_def=true
+
+# Control overwrite behavior
+export conductor.worker.all.overwrite_task_def=false  # Don't overwrite existing
+
+# Enable strict schema validation
+export conductor.worker.process_order.strict_schema=true  # No extra fields allowed
+```
+
+### Schema Validation Modes
+
+The `strict_schema` flag controls JSON Schema validation strictness:
+
+**Lenient Mode (default, strict_schema=False):**
+```json
+{
+  "type": "object",
+  "properties": {...},
+  "additionalProperties": true  ← Extra fields allowed
+}
+```
+
+**Strict Mode (strict_schema=True):**
+```json
+{
+  "type": "object",
+  "properties": {...},
+  "additionalProperties": false  ← Extra fields rejected
+}
+```
+
+**Use Cases:**
+- **Lenient (default):** Development, backward compatibility, flexible integrations
+- **Strict:** Production, strict validation, contract enforcement
+
+**Example:**
+```python
+@worker_task(
+    task_definition_name='validate_order',
+    register_task_def=True,
+    strict_schema=True  # Enforce strict validation
+)
+def validate_order(order_id: str, amount: float) -> dict:
+    return {}
+
+# Generated schema will reject inputs with extra fields
+```
+
+### Task Definition Overwrite Control
+
+The `overwrite_task_def` flag controls update behavior:
+
+**Overwrite Mode (default, overwrite_task_def=True):**
+- Always calls `update_task_def()` (overwrites existing)
+- Ensures server has latest configuration from code
+- Use in development or when task config changes frequently
+
+**No-Overwrite Mode (overwrite_task_def=False):**
+- Checks if task exists first
+- Only creates if doesn't exist
+- Preserves manual changes on server
+- Use when tasks are managed outside code
+
+**Example:**
+```python
+@worker_task(
+    task_definition_name='stable_task',
+    register_task_def=True,
+    overwrite_task_def=False  # Don't overwrite existing config
+)
+def stable_worker(data: dict) -> dict:
+    return {}
+
+# If task exists on server, keeps existing configuration
+# If task doesn't exist, registers new one
 ```
 
 ### Startup Configuration Logging
@@ -1435,7 +1511,7 @@ Expected improvements for I/O-bound async workers:
 
 ## Changelog
 
-### Version 4.1 (2025-11-28)
+### Version 4.1 (2025-11-30)
 - Enhanced Worker Discovery section with comprehensive WorkerLoader documentation
 - Expanded Long-Running Tasks section with detailed lease extension patterns
 - Added practical examples for checkpointing and external system polling
@@ -1451,16 +1527,26 @@ Expected improvements for I/O-bound async workers:
   - Generates JSON Schema (draft-07) from Python type hints
   - Supports dataclasses, Optional, List, Dict, Union types
   - Creates schemas named {task_name}_input and {task_name}_output
-  - Does not overwrite existing definitions or schemas
   - Works for both TaskRunner and AsyncTaskRunner
+  - Added task_def parameter for advanced configuration (retry, timeout, rate limits)
+- **Added overwrite_task_def and strict_schema flags:**
+  - `overwrite_task_def` (default: True) - Controls whether to overwrite existing task definitions
+  - `strict_schema` (default: False) - Controls additionalProperties in JSON schemas
+  - Both configurable via environment variables
+  - Applies to both sync and async workers
 - **Added TaskUpdateFailure event:**
   - Published when task update fails after all retry attempts (4 retries with exponential backoff: 10s/20s/30s)
   - Contains TaskResult for recovery/logging
   - Enables external handling of critical update failures
   - Event count: 7 total events (was 6)
+- **Fixed Optional[T] handling:**
+  - Optional[T] parameters are NOT required in schema
+  - Optional[T] parameters are marked nullable
+  - Works correctly with nested dataclasses
 - Added detailed polling loop with dynamic batch sizing examples
 - Improved troubleshooting guidance
 - Fixed class-based worker support in TaskHandler async detection
+- Fixed task_def_template passing in scan_for_annotated_workers path
 
 ### Version 4.0 (2025-11-28)
 - AsyncTaskRunner: Pure async/await execution (zero thread overhead)

examples/workers_e2e.py

@@ -20,6 +20,12 @@
   * Optional fields (Optional[str], default values)
   * Generates JSON Schema draft-07 automatically
   * Registers schemas as {task_name}_input and {task_name}_output
+- ⭐ CONFIGURATION FLAGS:
+  * overwrite_task_def: Control whether to overwrite existing task definitions
+  * strict_schema: Control JSON schema validation strictness (additionalProperties)
+  * Both flags support environment variable override
+  * Global: conductor.worker.all.<property>
+  * Worker-specific: conductor.worker.<task_name>.<property>
 
 Usage:
     export CONDUCTOR_SERVER_URL="http://localhost:8080/api"
@@ -85,7 +91,9 @@
     task_definition_name='calculate',
     thread_count=100,  # High concurrency - async workers can handle it!
     poll_timeout=10,
-    register_task_def=True
+    register_task_def=True,  # Auto-register task definition with JSON schema
+    overwrite_task_def=True,  # Always update definition on server (default: true)
+    strict_schema=False  # Allow additional properties in schema (default: false)
 )
 async def calculate_fibonacci(n: int) -> int:
     """
@@ -101,6 +109,11 @@ async def calculate_fibonacci(n: int) -> int:
     - Concurrency: Up to 100 concurrent tasks
     - Memory: ~3-6 MB per process
 
+    Configuration:
+    - register_task_def=True: Auto-registers task definition + JSON schema
+    - overwrite_task_def=True: Always updates server (default behavior)
+    - strict_schema=False: Allows extra fields in input (lenient, default)
+
     Note: This is a CPU-bound task (fibonacci calculation), which isn't
     ideal for async workers. Use this pattern for I/O-bound operations
     (HTTP calls, database queries, file I/O).
@@ -161,17 +174,18 @@ class OrderRequest:
 
 
 # Create TaskDef with advanced configuration for the complex order worker
+# This demonstrates using task_def parameter to specify retry, timeout, and rate limiting
 complex_order_task_def = TaskDef(
     name='process_complex_order',  # Will be overridden by task_definition_name
     description='Process customer orders with complex validation and retry logic',
     retry_count=3,                          # Retry up to 3 times on failure
     retry_logic='EXPONENTIAL_BACKOFF',      # Use exponential backoff between retries
     retry_delay_seconds=10,                 # Start with 10 second delay
-    backoff_scale_factor=3,                 # Double delay each retry (10s, 20s, 40s)
+    backoff_scale_factor=3,                 # Triple delay each retry (10s, 30s, 90s)
     timeout_seconds=600,                    # Task must complete within 10 minutes
     response_timeout_seconds=120,           # Each execution attempt has 2 minutes
     timeout_policy='RETRY',                 # Retry on timeout
-    concurrent_exec_limit=30,                # Max 5 concurrent executions
+    concurrent_exec_limit=30,               # Max 30 concurrent executions
     rate_limit_per_frequency=100,           # Max 100 executions
     rate_limit_frequency_in_seconds=60,     # Per 60 seconds
     poll_timeout_seconds=30                 # Long poll timeout for efficiency
@@ -180,8 +194,13 @@ class OrderRequest:
 @worker_task(
     task_definition_name='process_complex_order',
     thread_count=10,
-    register_task_def=True,  # Will auto-generate and register JSON schema!
-    task_def=complex_order_task_def  # Advanced task configuration
+    register_task_def=True,  # Auto-register task definition with JSON schemas
+    task_def=complex_order_task_def,  # Advanced task configuration (retry, timeout, rate limits)
+    overwrite_task_def=True,  # Always update task definition on server (default: true)
+    strict_schema=False  # Lenient validation - allows extra fields (default: false)
+    # Can be overridden via env:
+    #   export conductor.worker.process_complex_order.overwrite_task_def=false
+    #   export conductor.worker.process_complex_order.strict_schema=true
 )
 async def process_complex_order(
     order: OrderRequest,
@@ -203,19 +222,30 @@ async def process_complex_order(
        - Schema registered as: process_complex_order_input (v1)
 
     2. ADVANCED TASK CONFIGURATION via task_def parameter:
-       - Retry policy: 3 retries with EXPONENTIAL_BACKOFF (10s, 20s, 40s)
+       - Retry policy: 3 retries with EXPONENTIAL_BACKOFF (10s, 30s, 90s)
        - Timeouts: 10 min total, 2 min per execution
        - Rate limiting: Max 100 executions per 60 seconds
-       - Concurrency: Max 5 concurrent executions
+       - Concurrency: Max 30 concurrent executions
        - All configured via TaskDef object passed to @worker_task
 
+    3. CONFIGURATION FLAGS:
+       - overwrite_task_def=True: Always updates task definition on server
+         * Use False to preserve manual server changes
+         * Configurable: export conductor.worker.process_complex_order.overwrite_task_def=false
+
+       - strict_schema=False: Lenient JSON schema validation (allows extra fields)
+         * Use True for strict validation (rejects extra fields)
+         * Configurable: export conductor.worker.process_complex_order.strict_schema=true
+
     Benefits:
     - Input validation in Conductor UI
     - Type-safe workflow design
     - Auto-completion in workflow editor
     - Runtime validation of task inputs
     - Production-ready retry and timeout policies
     - Rate limiting to protect downstream services
+    - Flexible schema validation modes
+    - Environment-based configuration control
     """
     # Simulate order processing
     ctx = get_task_context()