---
name: gdtest-skill-complex
description: >
  Orchestrate recurring, cron-scheduled, and one-shot tasks with
  gdtest-skill-complex. Supports worker pools, task chaining,
  cron expressions, timeouts, and fire-and-forget patterns.
license: MIT
compatibility: Requires Python >=3.10.
metadata:
  author: gdg-test-suite
  version: "3.1"
  tags:
    - task-scheduling
    - cron
    - orchestration
    - worker-pool
---

# gdtest-skill-complex

A task-orchestration framework for scheduling, chaining, and
monitoring background work.

## Quick start

```python
from gdtest_skill_complex import Scheduler

sched = Scheduler(workers=4)
sched.every(seconds=60, fn=check_health, name="healthcheck")
sched.cron("0 2 * * *", fn=nightly_backup, name="backup")
sched.run()
```

## Skill directory structure

This skill ships with companion files for agent consumption:

```
skills/gdtest-skill-complex/
├── SKILL.md              ← This file
├── references/
│   ├── api-cheatsheet.md ← Quick-reference for all public APIs
│   └── migration-v2-v3.md ← Migration guide from v2 to v3
├── scripts/
│   ├── setup-env.sh      ← Environment bootstrap script
│   └── run-tests.sh      ← Test runner with coverage
└── assets/
    └── config-template.yaml ← Starter configuration file
```

## When to use what

| Need | Use |
|------|-----|
| Fixed-interval polling | `scheduler.every(seconds=30, fn=poll)` |
| Cron-scheduled jobs | `scheduler.cron('*/5 * * * *', fn=job)` |
| One-shot delayed task | `scheduler.once(delay=120, fn=migrate)` |
| Sequential pipeline | `scheduler.chain(extract, transform, load)` |
| Fire-and-forget | `scheduler.submit(fn=send_email)` |
| Concurrent workers | `Scheduler(workers=cpu_count())` |

## Core concepts

### Scheduler

The `Scheduler` manages a pool of workers and a task registry.
Tasks are added via `.every()`, `.cron()`, `.once()`, or `.submit()`.
Call `.run()` to start blocking execution, or `.stop()` to shut down
gracefully.

### Task

A `Task` wraps a callable with a unique name and optional timeout.
Task names **must be unique** within a scheduler — duplicates raise
`DuplicateTaskError`.

### CronExpr

Parses and evaluates standard 5-field cron expressions
(minute, hour, day-of-month, month, day-of-week). Does **not**
support second-level granularity.

```python
from gdtest_skill_complex import CronExpr

expr = CronExpr("*/5 * * * *")   # every 5 minutes
expr.matches(datetime.now())       # True/False
```

### TaskResult

Returned by `.run()` and `.chain()`. Contains the execution status
(`pending`, `success`, `failed`, `timeout`), the return value, and
any exception.

## Task chaining

Chain tasks to build sequential pipelines where each task's output
feeds the next:

```python
extract = Task("extract", fn=extract_data)
transform = Task("transform", fn=clean_and_normalize)
load = Task("load", fn=write_to_db)

results = scheduler.chain(extract, transform, load)
for r in results:
    print(f"{r.task.name}: {r.status}")
```

## Reference files

### API cheatsheet (`references/api-cheatsheet.md`)

A condensed reference of every public class and function:

| Symbol | Signature | Purpose |
|--------|-----------|---------|
| `Scheduler` | `(workers=1, name='default')` | Central orchestrator |
| `Task` | `(name, fn=None, timeout=None)` | Unit of work |
| `CronExpr` | `(expr)` | Cron parser |
| `TaskResult` | `(task, status, value, error)` | Execution result |
| `every()` | `(seconds, fn)` | Module-level interval shortcut |
| `cron()` | `(expr, fn)` | Module-level cron shortcut |
| `once()` | `(delay, fn)` | Module-level one-shot shortcut |
| `chain()` | `(*tasks)` | Module-level chain shortcut |

### Migration guide (`references/migration-v2-v3.md`)

Key changes from v2 to v3:

1. `Scheduler.interval()` renamed to `Scheduler.every()`.
2. `Task.callback` renamed to `Task.fn`.
3. `CronExpr` now validates on construction (was lazy).
4. Worker count defaults to 1 (was `os.cpu_count()`).

## Scripts

### `scripts/setup-env.sh`

Bootstrap a development environment:

```bash
#!/usr/bin/env bash
set -euo pipefail
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev,test]"
echo "Environment ready."
```

### `scripts/run-tests.sh`

Run the test suite with coverage:

```bash
#!/usr/bin/env bash
set -euo pipefail
pytest tests/ --cov=gdtest_skill_complex --cov-report=term-missing
```

## Configuration template

The `assets/config-template.yaml` provides a starter configuration:

```yaml
# gdtest-skill-complex configuration
scheduler:
  workers: 4
  name: production

tasks:
  - name: healthcheck
    type: every
    seconds: 30
    fn: app.health.check

  - name: nightly-backup
    type: cron
    expr: "0 2 * * *"
    fn: app.backup.run
    timeout: 3600

  - name: weekly-report
    type: cron
    expr: "0 9 * * 1"
    fn: app.reports.weekly
```

## Error handling

```python
results = scheduler.run()
for r in results:
    if r.status == "failed":
        print(f"Task {r.task.name} failed: {r.error}")
    elif r.status == "timeout":
        print(f"Task {r.task.name} timed out after {r.task.timeout}s")
```

## Capabilities and boundaries

**What agents can configure:**

- Create schedulers with worker pools
- Register interval, cron, one-shot, and fire-and-forget tasks
- Chain tasks into sequential pipelines
- Set per-task timeouts
- Parse and validate cron expressions
- Use reference files for quick API lookups
- Run setup and test scripts

**Requires human setup:**

- Deploying as a system service (systemd, Docker, etc.)
- Configuring monitoring and alerting
- Setting up log aggregation
- Database and credential provisioning

## Resources

- [llms.txt](llms.txt) — Indexed API reference for LLMs
- [llms-full.txt](llms-full.txt) — Comprehensive documentation for LLMs