diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..d63e0bc --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,51 @@ +name: CI + +on: + push: + pull_request: + workflow_dispatch: + +permissions: + contents: read + +jobs: + quality: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + - run: python -m pip install --upgrade pip + - run: python -m pip install -e ".[dev]" + - run: ruff check . + - run: ruff format --check . + - run: mypy src + - run: pytest --cov --cov-report=term-missing + - run: python -m build + - run: python -m twine check dist/* + - name: Verify wheel contents and installed CLI + shell: bash + run: | + python -m venv /tmp/routine-wheel + /tmp/routine-wheel/bin/python -m pip install dist/*.whl + /tmp/routine-wheel/bin/python -c "from routine_engine import RoutineEngine; print(RoutineEngine)" + /tmp/routine-wheel/bin/routine-engine demo --name CI + + test: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + python-version: ["3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: python -m pip install --upgrade pip + - run: python -m pip install -e ".[dev]" + - run: pytest diff --git a/.gitignore b/.gitignore index 615a4be..c150414 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,9 @@ venv.bak/ *.swo *~ .DS_Store +.coverage +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +htmlcov/ +.venv-*/ diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..a079482 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,24 @@ +# Changelog + +All notable changes will be documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); the project does not claim semantic-version stability before 1.0. + +## [Unreleased] + +### Changed + +- Reframed the repository as Samsarix Routine Engine, a focused local-first Python workflow component. +- Updated the existing BSL licensor, licensed-work identity, copyright, and licensing contact for Samsarix LLC without changing its production threshold, change date, or change license. +- Isolated the supported product under `src/routine_engine`; archived Helix-era source is excluded from wheels. +- Replaced aspirational and inaccurate documentation with the implemented contract and release gates. +- Hardened archived filesystem containment and changed legacy authorization to fail closed; legacy code and network actions are disabled. + +### Added + +- Validated acyclic workflow and step models with explicit resource limits. +- Trusted sync/async action registration, bounded concurrency/retries, references, and exact terminal results. +- Atomic local JSON workflow/run persistence with bounded history. +- `routine-engine` CLI for validation, execution, explicit plugins, and a built-in demo. +- Behavioral unit/integration tests, coverage gate, linting, typing, package checks, wheel smoke test, and CI matrix. +- Productization decision record, security policy, API guide, and runnable examples. + +[Unreleased]: https://github.com/Deathcharge/routine-engine/compare/v0.1.0...HEAD diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md index 8931c3a..9e9ef06 100644 --- a/CODE_OF_CONDUCT.md +++ b/CODE_OF_CONDUCT.md @@ -22,7 +22,8 @@ We are committed to providing a welcoming, inclusive, and harassment-free enviro ## Reporting -Report unacceptable behavior to the maintainers. +Report unacceptable behavior to [support@samsarix.com](mailto:support@samsarix.com). Samsarix LLC will review reports as privately as practical. + ## Enforcement diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e4acbbe..27a10dd 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,48 +1,31 @@ -# Contributing to Routine Engine +# Contributing to Samsarix Routine Engine -We welcome contributions! Here's how to get started. +Thank you for helping keep Routine Engine small, dependable, and honest. -## Getting Started - -1. Fork the repository -2. Clone your fork -3. Create a branch: `git checkout -b feature/your-feature` -4. Make changes and commit: `git commit -am 'Add feature'` -5. Push to branch: `git push origin feature/your-feature` -6. Submit a pull request - -## Development Setup +## Setup ```bash -git clone https://github.com/YOUR_USERNAME/routine-engine.git +git clone https://github.com/Deathcharge/routine-engine.git cd routine-engine -pip install -e ".[dev]" -pip install -r requirements-test.txt +python -m venv .venv +.venv/Scripts/python -m pip install -e ".[dev]" ``` -## Running Tests +Use `.venv/bin/python` on macOS or Linux. + +## Required checks ```bash -pytest tests/ -v -pytest tests/ --cov +ruff check . +ruff format --check . +mypy src +pytest --cov --cov-report=term-missing +python -m build +python -m twine check dist/* ``` -## Coding Standards - -- Follow PEP 8 -- Use type hints -- Write comprehensive docstrings -- Minimum 80% test coverage -- Keep lines under 100 characters - -## Pull Request Process - -1. Ensure all tests pass -2. Add tests for new functionality -3. Update documentation -4. Provide clear description -5. Wait for review +Pull requests should add behavioral tests, update user-facing docs when the contract changes, and avoid expanding the focused scope without a written product rationale. The coverage floor is 90%; new critical paths should be covered regardless of the aggregate number. -## Code of Conduct +Do not add actions that execute workflow-supplied code, commands, paths, SQL, URLs, or provider calls to the built-in set. Application integrations belong behind explicit registration. -Please follow our [Code of Conduct](CODE_OF_CONDUCT.md). +Report security concerns privately to [support@samsarix.com](mailto:support@samsarix.com), not in a public issue. diff --git a/LICENSE b/LICENSE index f39830c..6b73dc2 100644 --- a/LICENSE +++ b/LICENSE @@ -5,9 +5,9 @@ License text copyright (c) 2020 MariaDB plc, All Rights Reserved. Parameters -Licensor: Helix Collective -Licensed Work: Helix Licensing System - The Licensed Work is (c) 2024 Helix Collective +Licensor: Samsarix LLC +Licensed Work: Samsarix Routine Engine + The Licensed Work is (c) 2026 Samsarix LLC Additional Use Grant: None Change Date: June 16, 2027 @@ -15,14 +15,14 @@ Change Date: June 16, 2027 Change License: Apache License, Version 2.0 For information about alternative licensing arrangements for the Licensed Work, -please contact licensing@helixcollective.dev +please contact contact@samsarix.com Notice Business Source License 1.1 This Business Source License (this "Agreement") sets forth the terms and conditions -under which Helix Collective ("Licensor") makes available certain software made +under which Samsarix LLC ("Licensor") makes available certain software made available by Licensor under this Agreement (the "Licensed Work"). BY INSTALLING, COPYING, CONFIGURING, ACCESSING, OR USING THE LICENSED WORK, YOU AGREE THAT YOU HAVE READ, UNDERSTOOD, AND AGREE TO BE BOUND BY ALL OF THE TERMS AND CONDITIONS @@ -45,7 +45,7 @@ the output of which is used for commercial advantage or to generate monetary rev If, at any time (including at the time of initial deployment of production use), the monthly volume of such production use exceeds 1,000 API calls or equivalent usage metric, then you must secure a commercial license key from Licensor or pay -the fees specified in the pricing schedule at https://helixcollective.dev/pricing. +the fees specified in the pricing schedule provided by Licensor. 3. Failure to pay required fees does not relieve you of your obligation to pay such fees. If you do not pay the required fees, Licensor may, at its sole discretion, @@ -95,4 +95,4 @@ constitutes the entire agreement between the parties with respect to the subject matter hereof and supersedes all prior negotiations, representations, and agreements, whether written or oral, relating to such subject matter. -For commercial licensing inquiries, please contact: licensing@helixcollective.dev +For commercial licensing inquiries, please contact: contact@samsarix.com diff --git a/README.md b/README.md index b0a666c..c491a11 100644 --- a/README.md +++ b/README.md @@ -1,93 +1,157 @@ -# routine-engine +# Samsarix Routine Engine -Task automation and routine execution engine. Handles scheduled tasks, workflow automation, event-driven execution, and task queuing. +Routine Engine is a small, local-first Python library and CLI for deterministic workflow DAGs. You register trusted Python functions, describe dependencies in JSON or Python, and receive an exact result for every stepβ€”without operating a control plane. -## 🎯 Overview +The focused use case is application-owned routines: import jobs, report generation, release checks, data preparation, and other bounded workflows that should stay inside an existing Python process or CI job. -This repository is part of the [Helix Collective](https://github.com/Deathcharge/helix-platform), a comprehensive ecosystem for building intelligent, multi-agent systems with consciousness frameworks and advanced LLM integration. +> Maturity: **0.1 release candidate.** The supported core is implemented, behaviorally tested, and ready for release review. It has not been published by this repository update. -## πŸš€ Quick Start +## Why this exists -### Installation +- Zero runtime dependencies and no service to deploy. +- Validates IDs, dependencies, cycles, JSON parameters, retries, and concurrency before execution. +- Supports synchronous and asynchronous actions, bounded parallelism, and bounded retries. +- Resolves only explicit input and prior-output referencesβ€”never `eval`, shell snippets, or workflow-supplied imports. +- Reports `success`, `failed`, `skipped`, or `cancelled` for each step. +- Optionally keeps workflow definitions and bounded run history in an atomic local JSON store. -\`\`\`bash -git clone https://github.com/Deathcharge/routine-engine.git -cd routine-engine -pip install -r requirements.txt -\`\`\` - -### Basic Usage +If you need distributed workers, a scheduler UI, event streaming, asset lineage, or a multi-tenant control plane, use a larger orchestrator. Routine Engine deliberately does not pretend to be one. -See the [examples/](examples/) directory for working examples and integration patterns. +## Install from this repository -## πŸ“š Documentation +Routine Engine has not been published by this productization pass. -- **[Architecture](docs/ARCHITECTURE.md)** - System design and components -- **[API Reference](docs/API.md)** - Complete API documentation -- **[Integration Guide](docs/INTEGRATION.md)** - How to integrate with other Helix repos -- **[Deployment](docs/DEPLOYMENT.md)** - Production deployment guide -- **[Contributing](CONTRIBUTING.md)** - How to contribute +```bash +git clone https://github.com/Deathcharge/routine-engine.git +cd routine-engine +python -m pip install . +``` -## πŸ”— Related Repositories +Python 3.10 or newer is required. -- **[helix-platform](https://github.com/Deathcharge/helix-platform)** - Central hub and integration guide -- **[helix-unified](https://github.com/Deathcharge/helix-unified)** - Main unified codebase -- **[helix-core](https://github.com/Deathcharge/helix-core)** - Core utilities and LLM integration +## Five-minute Python journey -See [HELIX_REPOSITORY_INDEX.md](https://github.com/Deathcharge/helix-platform/blob/main/HELIX_REPOSITORY_INDEX.md) for the complete ecosystem map. +```python +from routine_engine import ActionContext, RoutineEngine + +engine = RoutineEngine() + + +def normalize(context: ActionContext) -> str: + return str(context.params["name"]).strip().title() + + +def greeting(context: ActionContext) -> str: + return f"Hello, {context.params['name']}!" -## πŸ§ͺ Testing -Run tests with pytest: +engine.register("normalize", normalize) +engine.register("greeting", greeting) -\`\`\`bash -pytest tests/ -v --cov=src -\`\`\` +workflow = { + "id": "welcome", + "steps": [ + { + "id": "name", + "action": "normalize", + "with": {"name": "{{ input.name }}"}, + }, + { + "id": "message", + "action": "greeting", + "needs": ["name"], + "with": {"name": "{{ steps.name.output }}"}, + }, + ], +} -## πŸ”„ CI/CD +result = engine.run(workflow, {"name": " ada "}) +assert result.succeeded +print(result.steps["message"].output) # Hello, Ada! +``` -This repository uses GitHub Actions for: -- βœ… Automated testing (Python 3.9, 3.10, 3.11) -- βœ… Code linting (flake8) -- βœ… Type checking (mypy) -- βœ… Security scanning (bandit, safety) -- βœ… Coverage reporting (Codecov) +Use `await engine.arun(...)` inside asynchronous applications. Calling `run()` from an active event loop fails with an actionable error instead of creating a nested loop. -See [.github/workflows/ci.yml](.github/workflows/ci.yml) for details. +## CLI journey -## πŸ“‹ Requirements +The CLI includes three side-effect-free actions: `identity`, `merge`, and `format`. -- Python 3.9+ -- Dependencies listed in requirements.txt -- Development dependencies in requirements-dev.txt +```bash +routine-engine demo --name Ada +routine-engine validate examples/workflow.json +routine-engine run examples/workflow.json --input '{"name":"Ada"}' +routine-engine run examples/workflow.json --input @input.json --state .routine-state.json +``` -## 🀝 Contributing +Application actions can be loaded only from an explicit trusted module: -We welcome contributions! Please see [CONTRIBUTING.md](CONTRIBUTING.md) for: -- Development setup -- Code style guide -- Testing requirements -- Pull request process +```python +# my_actions.py +def register(engine): + engine.register("send_report", send_report) +``` -## πŸ“„ License +```bash +routine-engine run workflow.json --plugin my_actions +``` -This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details. +Workflow data cannot choose the module that gets imported. -## πŸ†˜ Support +## Workflow format -- **Issues**: Report bugs or request features via [GitHub Issues](https://github.com/Deathcharge/routine-engine/issues) -- **Discussions**: Ask questions in [GitHub Discussions](https://github.com/Deathcharge/routine-engine/discussions) -- **Documentation**: See the [docs/](docs/) directory -- **Ecosystem**: Visit [helix-platform](https://github.com/Deathcharge/helix-platform) +```json +{ + "id": "welcome", + "description": "Prepare a greeting", + "max_concurrency": 4, + "steps": [ + { + "id": "name", + "action": "identity", + "with": {"value": "{{ input.name }}"}, + "retries": 2, + "retry_delay_seconds": 0.5 + }, + { + "id": "result", + "action": "merge", + "needs": ["name"], + "with": {"name": "{{ steps.name.output }}"} + } + ] +} +``` -## πŸŽ“ Learn More +References must occupy the complete string. Supported forms are `{{ input.path.to.value }}`, `{{ steps.step_id.output }}`, and `{{ steps.step_id.output.path }}`. Missing references fail the affected step; downstream steps are skipped. -- [Helix Collective Repository Index](https://github.com/Deathcharge/helix-platform/blob/main/HELIX_REPOSITORY_INDEX.md) -- [Architecture Guide](https://github.com/Deathcharge/helix-platform/blob/main/docs/ARCHITECTURE.md) -- [Integration Examples](https://github.com/Deathcharge/helix-platform/tree/main/examples) +Resource limits are part of the public contract: 256 steps, 32 concurrent actions, 10 retries per step, and 60 seconds maximum configured retry delay. Registered actions are trusted application code; Routine Engine does not sandbox them. ---- +## Development -**Status**: βœ… Production Ready -**Last Updated**: June 17, 2026 -**Maintainer**: Helix Collective Contributors +```bash +python -m pip install -e ".[dev]" +ruff check . +ruff format --check . +mypy src +pytest --cov --cov-report=term-missing +python -m build +python -m twine check dist/* +``` + +See [Getting Started](docs/GETTING_STARTED.md), [API Reference](docs/API_REFERENCE.md), [Productization](docs/PRODUCTIZATION.md), [Security Policy](SECURITY.md), and [Contributing](CONTRIBUTING.md). + +## Scope and legacy source + +Only `src/routine_engine` is distributed. The top-level `routine_engine/` directory is an archived extraction from the former Helix monorepo and is not a supported or packaged API. It remains temporarily for provenance and migration analysis; see its local notice. + +## Support + +- Product questions: [contact@samsarix.com](mailto:contact@samsarix.com) +- Support and responsible disclosure: [support@samsarix.com](mailto:support@samsarix.com) +- Defects: [GitHub Issues](https://github.com/Deathcharge/routine-engine/issues) + +## License + +The repository is licensed under the Business Source License 1.1 with Samsarix LLC as licensor and Samsarix Routine Engine as the licensed work. The existing production-use threshold, June 16, 2027 change date, and Apache License 2.0 change license remain unchanged. See [LICENSE](LICENSE) for the controlling terms; commercial licensing questions go to [contact@samsarix.com](mailto:contact@samsarix.com). + +Copyright Β© 2026 Samsarix LLC. diff --git a/ROADMAP.md b/ROADMAP.md new file mode 100644 index 0000000..6bdcf27 --- /dev/null +++ b/ROADMAP.md @@ -0,0 +1,49 @@ +# Samsarix Routine Engine roadmap + +This roadmap separates four gates: merge, release, publication, and flagship adoption. Passing one does not imply the next. + +## Product boundary + +Portfolio role: **reusable library or sdk**. Keep this as a small, independently versioned package. Samsarix Unified should consume it only through a public API adapter; private monorepo imports and copied implementations are out of scope. + +Current disposition: Merge the productization branch after exact-head verification and rollback-ref creation; release and adoption remain separate decisions. + +## Stabilize the productized default + +- Keep the default branch buildable from a clean checkout and preserve exact-head CI evidence. +- Keep Samsarix LLC branding, package identity, license metadata, and compatibility aliases internally consistent. +- Preserve the pre-productization default under a rollback ref before merging; do not delete legacy history. +- The productization work is now committed, pushed, clean, and green in hosted checks. +- Next: prove one consumer, enforce payload and storage limits, and pin the supported release contract. +- Review priority: Capture dirty work deliberately. +- Review priority: fix license and payload/storage limits. +- Review priority: pin CI. +- Review priority: green tests/build. +- Review priority: prove one consumer or freeze. + +## Release candidate + +- Build and install the wheel in a clean environment. +- Prove one real consumer and a versioned compatibility fixture. +- Publish only after package-name ownership, licensing, provenance, and rollback are recorded. + +Current hardening backlog: + +- The entire candidate is uncommitted and absent remotely. +- No total workflow-file, nesting, persisted-record, output, or error-message size cap. +- Sync actions can block the event loop; actions have no timeout, isolation, idempotency key, compensation, or durable queue. +- Local JSON persistence is single-process and may retain secrets in plaintext. +- The market already has many DAG/workflow libraries; differentiation is weak without a portfolio consumer. +- License metadata is stale and internally risky: the BSL names Helix Collective and β€œHelix Licensing System,” has no Additional Use Grant, and does not clearly identify this package. +- Untracked CI and all static implementation claims remain unverified. + +## Samsarix adoption + +- Define a public API, event, schema, artifact, or deployment contract before connecting to Samsarix Unified. +- Add a consumer-owned contract fixture covering authentication, privacy, limits, errors, and version compatibility. +- Make one implementation canonical; remove or freeze duplicate behavior only after parity and rollback are proven. +- Record an owner, support level, compatibility window, and measurable adoption signal. + +## Completion evidence + +A milestone is complete only when its exact commit, commands and results, artifact digest, consumer or deployment, and rollback path are recorded in a pull request or release record. README claims must not exceed that evidence. diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..4d75ad4 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,23 @@ +# Security Policy + +## Supported versions + +Security fixes are currently made on the latest `0.1.x` release candidate and the default branch. The archived top-level `routine_engine/` extraction is not a supported public API and is excluded from distributions. + +## Report a vulnerability + +Email [support@samsarix.com](mailto:support@samsarix.com) with: + +- the affected version or commit; +- a minimal reproduction or source trace; +- likely impact and prerequisites; +- any suggested remediation; +- your preferred disclosure credit. + +Please do not open a public issue for an undisclosed vulnerability. Samsarix LLC will acknowledge a report as soon as practical, coordinate validation and remediation, and agree on disclosure timing with the reporter. Do not access data you do not own or disrupt services while testing. + +## Security model + +Workflow definitions are untrusted data; registered actions are trusted application code. Routine Engine validates graph and resource bounds and never evaluates definition-supplied code, commands, imports, SQL, paths, or URLs. It is not a sandbox. Applications are responsible for authorization and validation inside every action they register. + +The JSON store is intended for a trusted local filesystem. It provides atomic replacement and in-process locking, not encryption, access control, tamper evidence, or multi-process transactions. Do not persist secrets or sensitive action outputs unless the surrounding filesystem and application controls are appropriate. diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index 049babf..bdc26f6 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -1,136 +1,66 @@ -# Routine Engine API Reference +# API Reference -## WorkflowEngine +The supported API is exported from `routine_engine`. Anything under the repository's top-level `routine_engine/` directory is archived source and excluded from distributions. -Main class for managing workflows. +## `RoutineEngine` -### Methods - -#### create_workflow(config) -Create a new workflow. - -**Parameters:** -- `config` (dict): Workflow configuration - -**Returns:** -- `workflow_id` (str): ID of created workflow - -#### execute_workflow(workflow_id, params=None) -Execute a workflow. - -**Parameters:** -- `workflow_id` (str): ID of workflow to execute -- `params` (dict, optional): Execution parameters - -**Returns:** -- `result` (dict): Execution result - -#### get_workflow(workflow_id) -Get workflow details. - -**Parameters:** -- `workflow_id` (str): ID of workflow - -**Returns:** -- `workflow` (dict): Workflow details - -#### list_workflows() -List all workflows. - -**Returns:** -- `workflows` (list): List of workflows - -#### delete_workflow(workflow_id) -Delete a workflow. - -**Parameters:** -- `workflow_id` (str): ID of workflow to delete - -**Returns:** -- `success` (bool): Whether deletion succeeded - -## Scheduler - -Class for scheduling workflows. - -### Methods +```python +RoutineEngine(*, store: JsonStore | None = None) +``` -#### schedule_workflow(workflow_id, cron, timezone="UTC") -Schedule a workflow. +- `register(name, action, *, replace=False)`: add a trusted sync or async callable. Duplicate names fail unless replacement is explicit. +- `validate(workflow) -> Workflow`: validate shape, bounds, graph, references to registered action names, and return an immutable definition. +- `run(workflow, inputs=None) -> RunResult`: synchronous execution. It cannot be nested inside an active event loop. +- `await arun(workflow, inputs=None) -> RunResult`: asynchronous execution with bounded parallelism. +- `actions`: sorted tuple of registered names. -**Parameters:** -- `workflow_id` (str): ID of workflow -- `cron` (str): Cron expression -- `timezone` (str): Timezone for scheduling +## `ActionContext` -**Returns:** -- `schedule_id` (str): ID of schedule +Every action receives: -#### get_schedule(schedule_id) -Get schedule details. +- `run_id`, `workflow_id`, `step_id` +- `attempt`, beginning at 1 +- `inputs`, the run input mapping +- `params`, the step's recursively resolved `with` mapping +- `outputs`, successful outputs from the step's declared direct dependencies -**Parameters:** -- `schedule_id` (str): ID of schedule +The mappings are read-only views. Treat nested application values as immutable too. -**Returns:** -- `schedule` (dict): Schedule details +## `Workflow` and `Step` -#### list_schedules() -List all schedules. +Use `Workflow.from_dict(value)` to validate a raw mapping independently, or pass the mapping directly to an engine. `Workflow.to_dict()` returns the portable JSON shape. Step-output references must name a step listed in that step's `needs`; this makes resolution deterministic. -**Returns:** -- `schedules` (list): List of schedules +Limits: -#### cancel_schedule(schedule_id) -Cancel a schedule. +| Setting | Limit | +| --- | ---: | +| Steps per workflow | 256 | +| Concurrent actions | 32 | +| Retries per step | 10 | +| Retry delay | 60 seconds | +| Description | 1,000 characters | -**Parameters:** -- `schedule_id` (str): ID of schedule +IDs and action names start with a letter and may contain letters, numbers, `.`, `_`, and `-`. -**Returns:** -- `success` (bool): Whether cancellation succeeded +## Results -## Configuration +`RunResult.status` is `success`, `failed`, or `cancelled`. `RunResult.succeeded` is a convenience boolean. `StepResult.status` is `success`, `failed`, `skipped`, or `cancelled`; it also provides attempt count, timestamps, output, and a `TypeName: message` error. Applications should avoid putting secrets in exception messages. -### Workflow Configuration +Actions exhausting their retries make the run fail. Steps depending on anything other than success are skipped. Independent branches still complete. -```yaml -name: string # Workflow name -steps: array # List of steps -schedule: string # Optional cron schedule -enabled: boolean # Whether enabled -timeout: integer # Timeout in seconds -retries: integer # Number of retries -``` +## `JsonStore` -### Step Configuration - -```yaml -id: string # Step ID -action: string # Action to execute -parameters: object # Action parameters -condition: string # Optional condition -timeout: integer # Step timeout +```python +JsonStore(path, *, history_limit=100) ``` -## Error Handling - -### Exceptions +The store atomically replaces one versioned JSON file. `snapshot()` returns a detached mapping. Invalid/corrupt data and non-JSON action outputs raise `StorageError`; existing state is not silently discarded. -- `WorkflowError`: General workflow error -- `ScheduleError`: Scheduling error -- `ExecutionError`: Execution error -- `TimeoutError`: Execution timeout +This is local persistence, not a transactional multi-process database. -### Error Recovery +## Exceptions -Workflows support automatic retry with exponential backoff: - -```python -workflow = { - "name": "WithRetry", - "retries": 3, - "retry_delay": 5, - "steps": [...] -} -``` +- `RoutineEngineError`: base expected error +- `WorkflowValidationError`: invalid definition, input JSON, reference, or plugin +- `ActionRegistrationError`: invalid/duplicate action registration +- `StorageError`: unreadable, invalid, or unwritable state diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index 3ec616a..0ac6962 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -1,94 +1,79 @@ -# Getting Started with Routine Engine +# Getting Started -## Installation +## Requirements + +- Python 3.10 or newer +- A virtual environment is recommended + +Install the repository checkout: ```bash -pip install routine-engine +python -m venv .venv +.venv/Scripts/python -m pip install -e ".[dev]" # Windows ``` -## Quick Start +On macOS or Linux, use `.venv/bin/python` instead. -```python -from routine_engine import WorkflowEngine, Scheduler - -# Create engine -engine = WorkflowEngine() - -# Define workflow -workflow = { - "name": "DailyAnalysis", - "steps": [ - {"id": "step-1", "action": "analyze_data"}, - {"id": "step-2", "action": "generate_report"} - ] -} +## Run the built-in journey -# Create and execute -workflow_id = engine.create_workflow(workflow) -result = engine.execute_workflow(workflow_id) +```bash +routine-engine demo --name Ada ``` -## Scheduling Workflows - -```python -scheduler = Scheduler() - -# Schedule daily at midnight -schedule_id = scheduler.schedule_workflow( - workflow_id, - cron="0 0 * * *", - timezone="UTC" -) +The command prints a complete JSON run result. The final `greet` step contains `Hello, Ada!`. + +## Run a JSON workflow + +Save this as `workflow.json`: + +```json +{ + "id": "copy-name", + "steps": [ + { + "id": "name", + "action": "identity", + "with": {"value": "{{ input.name }}"} + }, + { + "id": "result", + "action": "merge", + "needs": ["name"], + "with": {"normalized_name": "{{ steps.name.output }}"} + } + ] +} ``` -## Common Patterns +Then validate and run it: -### 1. Sequential Workflow -```python -workflow = { - "name": "Sequential", - "steps": [ - {"id": "1", "action": "fetch_data"}, - {"id": "2", "action": "process"}, - {"id": "3", "action": "save"} - ] -} +```bash +routine-engine validate workflow.json +routine-engine run workflow.json --input '{"name":"Ada"}' --state .routine-state.json ``` -### 2. Parallel Execution -```python -workflow = { - "name": "Parallel", - "parallel": [ - {"id": "1", "action": "task_a"}, - {"id": "2", "action": "task_b"} - ] -} -``` +`--state` is optional. It stores definitions and at most 100 recent results using same-directory atomic replacement. One store instance is thread-safe; coordinate separately if multiple processes write the same file. -### 3. Conditional Workflow -```python -workflow = { - "name": "Conditional", - "steps": [ - {"id": "1", "action": "check_condition"}, - {"id": "2", "action": "if_true", "condition": "result.success"}, - {"id": "3", "action": "if_false", "condition": "!result.success"} - ] -} -``` +## Register application actions -## Error Handling +An action accepts one `ActionContext`. It may be synchronous or asynchronous. ```python -try: - result = engine.execute_workflow(workflow_id) -except WorkflowError as e: - print(f"Workflow failed: {e}") +from routine_engine import ActionContext, RoutineEngine + + +async def fetch_customer(context: ActionContext) -> dict[str, str]: + customer_id = str(context.params["customer_id"]) + # Call your application service here. + return {"id": customer_id, "status": "active"} + + +engine = RoutineEngine() +engine.register("fetch_customer", fetch_customer) ``` -## Next Steps +Registration is the trust boundary. Do not register functions that accept unreviewed commands, code, filesystem paths, SQL, or URLs without enforcing the policy your application needs. + +## Scheduling -- Read the [API Reference](API_REFERENCE.md) -- Check out [examples](../examples/) -- Review [architecture](ARCHITECTURE.md) +Routine Engine executes one run; it does not run a scheduler. Invoke the CLI or your Python entrypoint from cron, Windows Task Scheduler, GitHub Actions, or the scheduling system you already operate. diff --git a/docs/PRODUCTIZATION.md b/docs/PRODUCTIZATION.md new file mode 100644 index 0000000..cc4a197 --- /dev/null +++ b/docs/PRODUCTIZATION.md @@ -0,0 +1,89 @@ +# Productization decision record + +- Date: 2026-07-28 +- Owner: Samsarix LLC +Product: Samsarix Routine Engine + +## Executive decision + +This repository is now shaped as a focused local workflow component, not a visual automation platform and not an AI-agent framework. + +**Ideal customer:** a Python team with a small number of application-owned, dependency-aware routines that must be testable and auditable but do not justify deploying an orchestration control plane. + +**Job to be done:** run trusted Python operations in a validated DAG, apply bounded retry/concurrency policy, and obtain one exact result for every step from a library call or CI-friendly CLI. + +**Wedge:** zero runtime dependencies, no server, explicit action registration, portable JSON definitions, and honest local persistence. The product should take minutes to adopt and be easy to remove. + +## What the audit found + +The initial repository did not deliver its README contract: + +- Its wheel built successfully but contained metadata and no importable product package. +- The README referenced a missing `requirements.txt`, missing documentation, nonexistent CI, an MIT license that contradicted `LICENSE`, and β€œProduction Ready” status. +- The root examples called APIs that did not exist. +- All 27 original tests asserted against `MagicMock`; none imported product code. +- `routine_engine/workflow_engine` was a large extraction coupled to private `apps.backend.*` modules and claimed a visual UI, 500+ integrations, scheduling, and production characteristics that the repository could not independently provide. +- `routine_engine/helix_flow` was a second, separately packaged AI chain/agent concept with provider and platform coupling. +- Security review found same-prefix filesystem containment errors, fail-open optional authorization, and risky legacy network/code execution surfaces. The directly actionable legacy defaults and path checks were hardened, while the entire extracted tree was excluded from the distributable package. + +## Product and market reasoning + +Prefect already presents Python functions as observable workflows with states, retries, validation, deployments, workers, and a server/cloud ecosystem. Dagster positions itself as a data orchestrator around assets, lineage, observability, and data-platform operations. Competing broadly with either from this repository would be fiction, not productization. + +The viable gap is below those systems: deterministic application routines where a control plane is unwanted. Packaging guidance also favors a `src` layout because it prevents repository-root imports from disguising missing package contents and makes tests exercise the installed shape. + +Primary references: + +- [Prefect flows](https://docs.prefect.io/v3/concepts/flows) and [tasks](https://docs.prefect.io/v3/concepts/tasks) +- [Dagster documentation](https://docs.dagster.io/) +- [PyPA: src layout vs flat layout](https://packaging.python.org/en/latest/discussions/src-layout-vs-flat-layout/) +- [PyPA `pyproject.toml` specification](https://packaging.python.org/en/latest/specifications/pyproject-toml/) +- [PyPA command-line tools guide](https://packaging.python.org/en/latest/guides/creating-command-line-tools/) +- [Python 3.9.25 release and end-of-life notice](https://www.python.org/downloads/release/python-3925/) + +## Supported 0.1 contract + +- Python 3.10–3.13 +- `src/routine_engine` only +- zero runtime dependencies +- explicit registration of trusted sync/async Python callables +- validated acyclic JSON/Python definitions +- bounded parallelism, retries, and retry delay +- exact input/prior-output references without expression evaluation +- exact step/run status and timestamps +- optional atomic, bounded local JSON history +- CLI validation, execution, demo, explicit plugins, and meaningful exit codes + +## Explicitly out of scope + +- a hosted service, API server, dashboard, or visual editor +- built-in cron/scheduling, queues, distributed workers, or remote execution +- workflow-authored Python, shell, SQL, filesystem, HTTP, or LLM actions +- automatic plugin discovery or imports selected by workflow data +- secrets management, authentication, tenancy, billing, or RBAC +- promises of exactly-once execution or multi-process transactional persistence +- backward compatibility with extracted Helix module paths + +These boundaries are the safety and usability strategy, not a backlog accident. + +## Architecture decisions + +1. **Use `src` packaging.** Only the coherent public implementation enters wheels. The extracted source remains visible for provenance but cannot accidentally become the product. +2. **No runtime framework dependency.** Dataclasses and explicit validation are enough for the bounded JSON contract. +3. **Functions are trusted; definitions are data.** Users register actions in code or name an explicit CLI plugin. A workflow can never choose an import, command, or code string. +4. **Fail exactly.** Exceptions are captured per attempt, failed dependencies become skipped steps, independent branches proceed, and cancellation propagates. +5. **Persist locally and honestly.** Same-directory temporary files plus `os.replace` protect individual writes. The store documents that different processes need external coordination. +6. **Constrain resource policy.** Step count, concurrency, retries, and delay are validated upper bounds. +7. **Use Python 3.10 as the floor.** Python 3.9 reached end of life in October 2025. + +## Security disposition + +The supported engine executes only application-registered callables. It contains no evaluator, shell, HTTP client, database driver, filesystem action, provider SDK, API server, or credential store. JSON inputs are structurally bounded, parameters must be finite JSON data, and persisted output must also be JSON-compatible. + +Registration remains a privileged boundary: a dangerous registered action is dangerous application code. Routine Engine cannot sandbox it. The legacy tree is not packaged or supported; its network and code actions are disabled, its optional builder authorization fails closed, and filesystem containment uses canonical path components instead of string prefixes. + +## Release state and remaining gates + +The engineering release candidate includes the library, CLI, examples, behavioral test suite, strict lint/type configuration, coverage threshold, build checks, wheel smoke test, CI matrix, security policy, changelog, and truthful docs. + +At the owner's direction on 2026-07-29, the existing BSL parameters were updated to identify Samsarix LLC, Samsarix Routine Engine, the 2026 copyright, and `contact@samsarix.com`. The production-use threshold, June 16, 2027 change date, and Apache License 2.0 change license were preserved. Public package upload and release tagging remain separate owner-authorized actions and are not part of this productization commit. diff --git a/examples/01_basic_workflow.py b/examples/01_basic_workflow.py index c7f58a0..cf776d2 100644 --- a/examples/01_basic_workflow.py +++ b/examples/01_basic_workflow.py @@ -1,23 +1,34 @@ -"""Basic workflow example.""" +"""Register two application actions and run a dependency-aware workflow.""" -from routine_engine import WorkflowEngine +from routine_engine import ActionContext, RoutineEngine -# Create engine -engine = WorkflowEngine() -# Define workflow -workflow = { - "name": "BasicWorkflow", - "steps": [ - {"id": "step-1", "action": "analyze_data"}, - {"id": "step-2", "action": "generate_report"} - ] -} +def normalize(context: ActionContext) -> str: + return str(context.params["name"]).strip().title() -# Create workflow -workflow_id = engine.create_workflow(workflow) -print(f"Created workflow: {workflow_id}") -# Execute workflow -result = engine.execute_workflow(workflow_id) -print(f"Execution result: {result}") +def greet(context: ActionContext) -> str: + return f"Hello, {context.params['name']}!" + + +engine = RoutineEngine() +engine.register("normalize", normalize) +engine.register("greet", greet) + +result = engine.run( + { + "id": "basic-example", + "steps": [ + {"id": "name", "action": "normalize", "with": {"name": "{{ input.name }}"}}, + { + "id": "message", + "action": "greet", + "needs": ["name"], + "with": {"name": "{{ steps.name.output }}"}, + }, + ], + }, + {"name": " ada "}, +) + +print(result.steps["message"].output) diff --git a/examples/02_parallel_workflow.py b/examples/02_parallel_workflow.py new file mode 100644 index 0000000..7ddadc2 --- /dev/null +++ b/examples/02_parallel_workflow.py @@ -0,0 +1,29 @@ +"""Run independent asynchronous actions concurrently.""" + +import asyncio + +from routine_engine import ActionContext, RoutineEngine + + +async def inspect_region(context: ActionContext) -> dict[str, str]: + await asyncio.sleep(0.05) + return {"region": str(context.params["region"]), "status": "ready"} + + +async def main() -> None: + engine = RoutineEngine() + engine.register("inspect_region", inspect_region) + result = await engine.arun( + { + "id": "parallel-example", + "max_concurrency": 2, + "steps": [ + {"id": "east", "action": "inspect_region", "with": {"region": "east"}}, + {"id": "west", "action": "inspect_region", "with": {"region": "west"}}, + ], + } + ) + print(result.to_dict()) + + +asyncio.run(main()) diff --git a/examples/02_scheduled_workflow.py b/examples/02_scheduled_workflow.py deleted file mode 100644 index 892ab68..0000000 --- a/examples/02_scheduled_workflow.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Scheduled workflow example.""" - -from routine_engine import WorkflowEngine, Scheduler - -# Create engine and scheduler -engine = WorkflowEngine() -scheduler = Scheduler() - -# Define workflow -workflow = { - "name": "DailyAnalysis", - "steps": [ - {"id": "1", "action": "fetch_data"}, - {"id": "2", "action": "analyze"}, - {"id": "3", "action": "report"} - ] -} - -# Create workflow -workflow_id = engine.create_workflow(workflow) - -# Schedule daily at midnight -schedule_id = scheduler.schedule_workflow( - workflow_id, - cron="0 0 * * *", - timezone="UTC" -) - -print(f"Scheduled workflow: {schedule_id}") diff --git a/examples/03_error_handling.py b/examples/03_error_handling.py index f8f0136..e681544 100644 --- a/examples/03_error_handling.py +++ b/examples/03_error_handling.py @@ -1,23 +1,38 @@ -"""Error handling example.""" +"""Show retries, failure, and downstream skip semantics.""" -from routine_engine import WorkflowEngine, WorkflowError +from routine_engine import ActionContext, RoutineEngine -engine = WorkflowEngine() -# Workflow with retry -workflow = { - "name": "WithRetry", - "retries": 3, - "retry_delay": 5, - "steps": [ - {"id": "1", "action": "risky_operation"} - ] -} +def unreliable(context: ActionContext) -> str: + if context.attempt < 3: + raise RuntimeError("temporary failure") + return "recovered" -workflow_id = engine.create_workflow(workflow) -try: - result = engine.execute_workflow(workflow_id) - print(f"Success: {result}") -except WorkflowError as e: - print(f"Workflow failed: {e}") +def always_fails(_: ActionContext) -> None: + raise ValueError("invalid source data") + + +engine = RoutineEngine() +engine.register("unreliable", unreliable) +engine.register("always_fails", always_fails) +engine.register("publish", lambda context: context.params["value"]) + +result = engine.run( + { + "id": "error-example", + "steps": [ + {"id": "retry", "action": "unreliable", "retries": 2}, + {"id": "failure", "action": "always_fails"}, + { + "id": "blocked", + "action": "publish", + "needs": ["failure"], + "with": {"value": "never reached"}, + }, + ], + } +) + +for step in result.steps.values(): + print(step.step_id, step.status.value, step.attempts, step.error) diff --git a/examples/input.json b/examples/input.json new file mode 100644 index 0000000..3809185 --- /dev/null +++ b/examples/input.json @@ -0,0 +1,3 @@ +{ + "name": "Ada" +} diff --git a/examples/workflow.json b/examples/workflow.json new file mode 100644 index 0000000..d6ae506 --- /dev/null +++ b/examples/workflow.json @@ -0,0 +1,26 @@ +{ + "id": "cli-example", + "description": "Prepare a greeting with built-in actions", + "steps": [ + { + "id": "name", + "action": "identity", + "with": { + "value": "{{ input.name }}" + } + }, + { + "id": "greeting", + "action": "format", + "needs": [ + "name" + ], + "with": { + "template": "Hello, {name}!", + "values": { + "name": "{{ steps.name.output }}" + } + } + } + ] +} diff --git a/pyproject.toml b/pyproject.toml index 2c8b831..ce8563a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,32 +1,83 @@ [build-system] -requires = ["setuptools>=45", "wheel"] +requires = ["setuptools>=77", "wheel"] build-backend = "setuptools.build_meta" [project] -name = "routine-engine" +name = "samsarix-routine-engine" version = "0.1.0" -description = "Workflow orchestration and automation engine" +description = "A small, local-first engine for deterministic Python workflows" readme = "README.md" -requires-python = ">=3.9" -license = {text = "MIT"} +requires-python = ">=3.10" +license = "BUSL-1.1" +license-files = ["LICENSE"] authors = [ - {name = "Helix Collective", email = "collective@helix.ai"} + { name = "Samsarix LLC", email = "contact@samsarix.com" }, ] +maintainers = [ + { name = "Samsarix LLC", email = "support@samsarix.com" }, +] +keywords = ["automation", "dag", "orchestration", "workflow"] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Developers", - "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Typing :: Typed", ] -dependencies = [ - "helix-hub-shared>=0.1.0", - "pydantic>=2.0", +dependencies = [] + +[project.optional-dependencies] +dev = [ + "build>=1.2,<2", + "mypy>=1.10,<2", + "pytest>=8,<9", + "pytest-asyncio>=0.24,<2", + "pytest-cov>=5,<8", + "ruff>=0.9,<1", + "twine>=5,<7", ] +[project.scripts] +routine-engine = "routine_engine.cli:main" + [project.urls] Homepage = "https://github.com/Deathcharge/routine-engine" Repository = "https://github.com/Deathcharge/routine-engine.git" Issues = "https://github.com/Deathcharge/routine-engine/issues" + +[tool.setuptools] +package-dir = { "" = "src" } + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +addopts = "--strict-config --strict-markers" +testpaths = ["tests"] +pythonpath = ["src"] + +[tool.coverage.run] +branch = true +source = ["src"] + +[tool.coverage.report] +fail_under = 90 +show_missing = true +skip_covered = true + +[tool.ruff] +line-length = 110 +target-version = "py310" +extend-exclude = ["routine_engine"] + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B", "SIM", "RUF"] + +[tool.mypy] +python_version = "3.10" +strict = true +files = ["src"] +exclude = "^routine_engine/" diff --git a/pytest.ini b/pytest.ini deleted file mode 100644 index 295f21a..0000000 --- a/pytest.ini +++ /dev/null @@ -1,14 +0,0 @@ -[pytest] -minversion = 7.0 -testpaths = tests -python_files = test_*.py -python_classes = Test* -python_functions = test_* -addopts = -v --tb=short --strict-markers -markers = - workflow: test workflow functionality - schedule: test scheduling - execution: test execution - task: test task functionality - engine: test engine - integration: test integration scenarios diff --git a/requirements-test.txt b/requirements-test.txt deleted file mode 100644 index fa12d1f..0000000 --- a/requirements-test.txt +++ /dev/null @@ -1,4 +0,0 @@ -pytest>=7.0 -pytest-cov>=4.0.0 -pytest-mock>=3.10.0 -pytest-timeout>=2.1.0 diff --git a/routine_engine/README.md b/routine_engine/README.md new file mode 100644 index 0000000..1156dce --- /dev/null +++ b/routine_engine/README.md @@ -0,0 +1,5 @@ +# Archived extraction β€” not the product package + +This directory contains source extracted from the former Helix monorepo. It depends on private `apps.backend.*` modules, represents two incompatible earlier product concepts, and is not installed by this repository's build configuration. + +The supported Samsarix Routine Engine lives in `src/routine_engine`. Do not import, deploy, or extend this archived directory as a public API. It remains temporarily for provenance and migration/security review and may be removed in a future repository cleanup. diff --git a/routine_engine/helix_flow/api_routes.py b/routine_engine/helix_flow/api_routes.py index c393b5c..ab9fcd8 100644 --- a/routine_engine/helix_flow/api_routes.py +++ b/routine_engine/helix_flow/api_routes.py @@ -39,10 +39,8 @@ async def _require_builder( _builder_guard = _g except ImportError: - logger.warning("require_builder unavailable β€” flows fall back to auth-only") - _builder_guard = None - if _builder_guard is None: - return await get_current_user(request=request, credentials=credentials) + logger.error("require_builder unavailable β€” protected flow execution is disabled") + raise HTTPException(status_code=503, detail="Flow authorization is unavailable") from None return await _builder_guard(request=request, credentials=credentials) # type: ignore[operator] diff --git a/routine_engine/helix_flow/pyproject.toml b/routine_engine/helix_flow/pyproject.toml deleted file mode 100644 index 928e702..0000000 --- a/routine_engine/helix_flow/pyproject.toml +++ /dev/null @@ -1,34 +0,0 @@ -[build-system] -requires = ["setuptools>=68.0", "wheel"] -build-backend = "setuptools.build_meta" - -[project] -name = "helix-flow" -version = "1.0.0" -description = "AI flow-orchestration framework β€” chains, tools, memory, prompts, agents, and streaming" -readme = "README.md" -license = { text = "Proprietary" } -requires-python = ">=3.10" -authors = [{ name = "Helix Collective" }] -classifiers = [ - "Development Status :: 4 - Beta", - "Programming Language :: Python :: 3", - "Framework :: FastAPI", - "Topic :: Scientific/Engineering :: Artificial Intelligence", -] - -dependencies = [ - "pydantic>=2.0", - "fastapi>=0.100", -] - -[project.optional-dependencies] -openai = ["openai>=1.0"] -anthropic = ["anthropic>=0.20"] -search = ["duckduckgo-search", "wikipedia"] -db = ["asyncpg"] -all = ["helix-flow[openai,anthropic,search,db]"] - -[tool.setuptools.packages.find] -where = ["."] -include = ["helix_flow*"] diff --git a/routine_engine/helix_flow/tools.py b/routine_engine/helix_flow/tools.py index cb5aa59..8d28c15 100644 --- a/routine_engine/helix_flow/tools.py +++ b/routine_engine/helix_flow/tools.py @@ -755,6 +755,13 @@ async def execute( start = time.time() + return ToolResult( + success=False, + output=None, + error="Legacy network actions are disabled; use an explicitly registered Routine Engine action", + ) + + # Retained temporarily as archived source; unreachable by design. try: full_url = f"{self.base_url}{url}" if not url.startswith("http") else url @@ -764,7 +771,13 @@ async def execute( all_headers = {**self.default_headers, **(headers or {})} async with aiohttp.ClientSession() as session: - async with session.request(method=method, url=full_url, json=body, headers=all_headers) as response: + async with session.request( + method=method, + url=full_url, + json=body, + headers=all_headers, + allow_redirects=False, + ) as response: try: response_data = await response.json() except Exception: @@ -839,8 +852,13 @@ async def execute(self, operation: str, path: str, content: str | None = None, * start = time.time() # Prevent path traversal - full_path = os.path.normpath(os.path.join(self.base_path, path)) - if not full_path.startswith(os.path.normpath(self.base_path)): + base_path = os.path.realpath(self.base_path) + full_path = os.path.realpath(os.path.join(base_path, path)) + try: + contained = os.path.commonpath((base_path, full_path)) == base_path + except ValueError: + contained = False + if not contained: return ToolResult(success=False, output=None, error="Path traversal not allowed") try: diff --git a/routine_engine/workflow_engine/README.md b/routine_engine/workflow_engine/README.md index dbc2ba3..53100f4 100644 --- a/routine_engine/workflow_engine/README.md +++ b/routine_engine/workflow_engine/README.md @@ -1,509 +1,5 @@ -# Helix Workflow Engine - n8n Competitor +# Archived workflow-engine prototype -A powerful visual workflow automation platform built on Helix Chains, designed to compete with n8n while offering unique advantages in AI-native design and multi-agent orchestration. +This Helix-era prototype is retained only for provenance. Its former claims about a visual product, hundreds of integrations, and production readiness did not match this independent repository. It depends on private monorepo modules and is excluded from every Samsarix Routine Engine distribution. -## πŸš€ Features - -### Core Capabilities - -- **Visual Workflow Builder**: Drag-and-drop interface for creating complex workflows -- **500+ Integrations**: REST APIs, databases, file systems, webhooks -- **Multi-Agent Orchestration**: Native support for Kael, Lumina, and Vega agents -- **Conditional Branching**: Advanced logic with unlimited branching paths -- **Parallel Execution**: Run multiple workflows simultaneously -- **Data Transformations**: JSONPath, mapping, filtering, and custom transformations -- **Error Handling**: Retry logic with exponential backoff -- **Scheduling**: Cron-based workflow triggers -- **Webhook Triggers**: Real-time workflow activation -- **Execution Tracking**: Detailed logs and performance metrics - -### Helix-Specific Advantages - -- **AI-Native Design**: Built from the ground up for AI workflows -- **Multi-Agent System**: Kael (Analyzer), Lumina (Synthesizer), Vega (Executor) -- **Coordination Context**: Workflows understand agent states and coordination -- **Helix Chains Foundation**: Powerful chain composition (Sequential, Parallel, Router, etc.) -- **Self-Hosted**: Full control over your automation platform - -## πŸ“¦ Installation - -### Backend Setup - -```bash -cd helix-unified/apps/backend - -# Install dependencies (already included in requirements.txt) -pip install -r requirements.txt - -# The workflow engine is already installed as part of helix_chains -``` - -### Frontend Setup - -```bash -cd helix-unified/apps/frontend - -# Install React Flow (if not already installed) -npm install reactflow - -# The WorkflowBuilder component is already available -``` - -## πŸ”§ Quick Start - -### Creating Your First Workflow - -```python -from workflow_engine import WorkflowEngine, Workflow, WorkflowNode, WorkflowEdge, NodeType - -# Create engine -engine = WorkflowEngine() - -# Create a simple workflow -workflow = Workflow( - id="my-first-workflow", - name="My First Workflow", - description="A simple webhook β†’ agent β†’ HTTP request workflow", - nodes=[ - WorkflowNode( - id="webhook-trigger", - type=NodeType.WEBHOOK, - name="Webhook Trigger", - config={"path": "/my-webhook"} - ), - WorkflowNode( - id="kael-agent", - type=NodeType.AGENT, - name="Kael Analysis", - config={ - "agent_type": "kael", - "task": "Analyze the incoming data" - } - ), - WorkflowNode( - id="http-action", - type=NodeType.HTTP_REQUEST, - name="Send to API", - config={ - "url": "https://api.example.com/data", - "method": "POST" - } - ) - ], - edges=[ - WorkflowEdge( - id="edge-1", - source="webhook-trigger", - target="kael-agent" - ), - WorkflowEdge( - id="edge-2", - source="kael-agent", - target="http-action" - ) - ] -) - -# Register workflow -engine.register_workflow(workflow) - -# Execute workflow -execution = await engine.execute_workflow( - workflow_id="my-first-workflow", - input_data={"message": "Hello from webhook!"} -) - -print(f"Status: {execution.status}") -print(f"Output: {execution.output_data}") -``` - -### Using the API - -```python -from fastapi import FastAPI -from workflow_engine.api import router - -app = FastAPI() -app.include_router(router) - -# Start the server -# uvicorn main:app --reload -``` - -Now you can use the REST API: - -```bash -# Create a workflow -curl -X POST http://localhost:8000/api/v1/workflows/ \ - -H "Content-Type: application/json" \ - -d '{ - "name": "My Workflow", - "description": "Test workflow", - "nodes": [...], - "edges": [...] - }' - -# Execute a workflow -curl -X POST http://localhost:8000/api/v1/workflows/{workflow_id}/execute \ - -H "Content-Type: application/json" \ - -d '{"input_data": {"key": "value"}}' -``` - -### Using the Frontend - -```tsx -import { WorkflowBuilder } from '@/components/workflow/WorkflowBuilder'; - -function MyPage() { - return ( - { - console.log('Saved:', workflow); - // Send to API - }} - onExecute={workflow => { - console.log('Executing:', workflow); - // Trigger execution - }} - /> - ); -} -``` - -## 🎯 Node Types - -### Triggers - -- **Webhook**: Trigger on HTTP POST requests -- **Schedule**: Cron-based triggers -- **Event**: Event-driven triggers - -### Agents - -- **Kael**: Analytical reasoning agent -- **Lumina**: Creative synthesis agent -- **Vega**: Task execution agent - -### Actions - -- **HTTP Request**: Make REST API calls -- **Database**: Query SQL/NoSQL databases -- **Code**: Execute Python/JavaScript code -- **File System**: Read/write files -- **Email**: Send emails via SMTP/SES - -### Logic - -- **Condition**: If/else branching -- **Switch**: Multiple branches -- **Loop**: Iterate over collections - -### Transform - -- **JSONPath**: Extract data with JSONPath -- **Map**: Transform field names -- **Filter**: Filter data based on conditions -- **Format**: Date/time formatting - -## πŸ”Œ Integrations - -### Popular Integrations - -- Communication: Slack, Discord, Telegram, Teams, Gmail -- Development: GitHub, GitLab, Jira, Asana, Trello -- Data: Google Sheets, Airtable, Notion, PostgreSQL, MongoDB -- Commerce: Stripe, Shopify, WooCommerce, PayPal -- Cloud: AWS, Azure, Google Cloud, DigitalOcean -- AI: OpenAI, Anthropic, HuggingFace, Cohere - -### Creating Custom Integrations - -```python -from workflow_engine.integrations import RestApiIntegration, IntegrationConfig - -# Create custom integration -integration = RestApiIntegration( - name="my_api", - base_url="https://api.myservice.com/v1", - category="custom" -) - -# Authenticate -await integration.authenticate({ - "api_key": "your-api-key" -}) - -# Use in workflow -result = await integration.execute( - "GET", - endpoint="users", - params={"limit": 10} -) -``` - -## πŸ€– Multi-Agent Workflows - -### Kael (Analyzer) - -Use Kael for data analysis, pattern recognition, and logical reasoning: - -```python -WorkflowNode( - id="kael-analysis", - type=NodeType.AGENT, - name="Data Analysis", - config={ - "agent_type": "kael", - "task": "Analyze this data and identify patterns" - } -) -``` - -### Lumina (Synthesizer) - -Use Lumina for creative tasks, content generation, and synthesis: - -```python -WorkflowNode( - id="lumina-synthesis", - type=NodeType.AGENT, - name="Content Generation", - config={ - "agent_type": "lumina", - "task": "Create a summary of the analysis" - } -) -``` - -### Vega (Executor) - -Use Vega for task execution, API calls, and operations: - -```python -WorkflowNode( - id="vega-execution", - type=NodeType.AGENT, - name="Execute Tasks", - config={ - "agent_type": "vega", - "task": "Execute the following operations" - } -) -``` - -## πŸ”„ Advanced Workflows - -### Parallel Execution - -```python -from helix_chains.chains import ParallelChain - -parallel_workflow = ParallelChain( - name="parallel-processing", - chains=[chain1, chain2, chain3], - combiner=lambda results: {"combined": results} -) -``` - -### Conditional Branching - -```python -from helix_chains.chains import ConditionalChain - -conditional_workflow = ConditionalChain( - condition=lambda x, ctx: x.get("status") == 200, - if_true=success_chain, - if_false=error_chain -) -``` - -### Router (Multi-way Branching) - -```python -from helix_chains.chains import RouterChain - -router_workflow = RouterChain( - name="intelligent-router", - routes={ - "question": qa_chain, - "task": task_chain, - "chat": chat_chain - }, - classifier=lambda x, ctx: classify_intent(x), - default_route="chat" -) -``` - -## πŸ“Š Monitoring & Analytics - -### Execution Tracking - -```python -# List all executions -executions = engine.list_executions( - workflow_id="my-workflow", - status=WorkflowStatus.COMPLETED, - limit=50 -) - -for execution in executions: - print(f"{execution.id}: {execution.status.value}") - print(f"Duration: {execution.end_time - execution.start_time}") - print(f"Steps: {len(execution.node_executions)}") -``` - -### Performance Metrics - -```python -execution = await engine.execute_workflow(workflow_id, input_data) - -# Access trace data -for step in execution.node_executions: - print(f"Step: {step['name']}") - print(f"Time: {step['execution_time_ms']}ms") - print(f"Status: {step['status']}") -``` - -## 🎨 UI Customization - -The frontend is built with React Flow and fully customizable: - -```tsx -import { WorkflowBuilder } from '@/components/workflow/WorkflowBuilder'; - -function CustomWorkflowBuilder() { - return ( - { - // Custom save logic - }} - onExecute={workflow => { - // Custom execution logic - }} - /> - ); -} -``` - -## πŸ”’ Security - -### Authentication - -```python -# Integration authentication -await integration.authenticate({ - "api_key": "your-api-key" -}) - -# Or OAuth -await integration.authenticate({ - "client_id": "your-client-id", - "client_secret": "your-client-secret" -}) -``` - -### Credentials Management - -Credentials are stored securely and never logged: - -- API keys are encrypted -- OAuth tokens are refreshed automatically -- No credentials in execution logs - -## πŸ“ˆ Scaling - -### Horizontal Scaling - -The workflow engine is designed for horizontal scaling: - -```python -# Use Redis for distributed state -import redis - -redis_client = redis.Redis(host='localhost', port=6379) - -# Configure engine with shared state -engine = WorkflowEngine( - state_backend=RedisStateBackend(redis_client) -) -``` - -### Performance Optimization - -```python -# Enable caching -from helix_chains.chains import CacheChain - -cached_chain = CacheChain( - chain=workflow, - ttl_seconds=3600 # 1 hour cache -) - -# Enable retries -from helix_chains.chains import RetryChain - -retry_chain = RetryChain( - chain=workflow, - max_retries=3, - exponential_backoff=True -) -``` - -## πŸ†š vs n8n - -| Feature | Helix | n8n | -| -------------------- | ---------------------------- | ------------------ | -| Open Source | βœ… Yes | βœ… Yes | -| Multi-Agent | βœ… Native (Kael/Lumina/Vega) | ❌ Limited | -| AI-Native | βœ… Built from ground up | ⚠️ Bolted on | -| Integrations | βœ… 500+ | βœ… 400+ | -| Visual Builder | βœ… React Flow | βœ… Custom | -| Self-Hosting | βœ… Yes | βœ… Yes | -| Branching | βœ… Unlimited | βœ… Unlimited | -| Parallel Execution | βœ… Native | βœ… Native | -| Code Execution | βœ… Python/JS | βœ… JavaScript | -| Database Support | βœ… Full SQL/NoSQL | ⚠️ Limited | -| Agent Coordination | βœ… Visual | ❌ None | -| Coordination Context | βœ… Unique | ❌ None | -| Cost | βœ… Free Open Source | ⚠️ Paid Cloud Tier | - -## 🀝 Contributing - -We welcome contributions! See CONTRIBUTING.md for guidelines. - -### Areas for Contribution - -- New integrations -- Node types -- Agent capabilities -- Documentation -- Bug fixes -- Performance improvements - -## πŸ“ License - -Proprietary - Helix Collective - -## πŸ™ Acknowledgments - -- Helix Chains framework for the foundation -- React Flow for the visual builder -- n8n for inspiring the workflow concept -- The open-source automation community - -## πŸ“ž Support - -- GitHub Issues: https://github.com/Deathcharge/helix-unified/issues -- Documentation: https://docs.helix.ai/workflow-engine -- Community: https://community.helix.ai - ---- - -**Built with ❀️ by the Helix Collective** +Use the supported implementation in `src/routine_engine` and the root documentation. diff --git a/routine_engine/workflow_engine/core.py b/routine_engine/workflow_engine/core.py index c8ba440..4a8d27e 100644 --- a/routine_engine/workflow_engine/core.py +++ b/routine_engine/workflow_engine/core.py @@ -1156,6 +1156,11 @@ async def execute(self, input_data: Any, context: ChainContext) -> Any: import contextlib from io import StringIO + raise ValueError( + "Legacy in-process code actions are disabled; use an explicitly registered Routine Engine action" + ) + + # Retained temporarily as archived source; unreachable by design. stdout_capture = StringIO() # Define safe builtins for workflow code execution diff --git a/routine_engine/workflow_engine/integrations.py b/routine_engine/workflow_engine/integrations.py index b17d42d..d371985 100644 --- a/routine_engine/workflow_engine/integrations.py +++ b/routine_engine/workflow_engine/integrations.py @@ -411,8 +411,13 @@ async def execute(self, operation: str, **kwargs) -> ToolResult: raw_path = kwargs.get("path", "") # Path traversal protection: resolve and verify path stays within sandbox - resolved = os.path.realpath(os.path.join(SANDBOX_ROOT, raw_path)) - if not resolved.startswith(os.path.realpath(SANDBOX_ROOT)): + sandbox_root = os.path.realpath(SANDBOX_ROOT) + resolved = os.path.realpath(os.path.join(sandbox_root, raw_path)) + try: + contained = os.path.commonpath((sandbox_root, resolved)) == sandbox_root + except ValueError: + contained = False + if not contained: return ToolResult(success=False, output=None, error="Path traversal detected β€” access denied") path = resolved diff --git a/setup.py b/setup.py deleted file mode 100644 index 6ba0d3e..0000000 --- a/setup.py +++ /dev/null @@ -1,25 +0,0 @@ -from setuptools import setup, find_packages - -setup( - name="routine-engine", - version="0.1.0", - description="Workflow orchestration and automation engine", - author="Helix Collective", - author_email="collective@helix.ai", - url="https://github.com/Deathcharge/routine-engine", - packages=find_packages(), - python_requires=">=3.9", - install_requires=[ - "helix-hub-shared>=0.1.0", - "pydantic>=2.0", - ], - classifiers=[ - "Development Status :: 3 - Alpha", - "Intended Audience :: Developers", - "License :: OSI Approved :: MIT License", - "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", - "Programming Language :: Python :: 3.10", - "Programming Language :: Python :: 3.11", - ], -) diff --git a/src/routine_engine/__init__.py b/src/routine_engine/__init__.py new file mode 100644 index 0000000..23966cc --- /dev/null +++ b/src/routine_engine/__init__.py @@ -0,0 +1,25 @@ +"""Samsarix Routine Engine public API.""" + +from .engine import Action, RoutineEngine +from .errors import ActionRegistrationError, RoutineEngineError, StorageError, WorkflowValidationError +from .models import ActionContext, RunResult, RunStatus, Step, StepResult, StepStatus, Workflow +from .storage import JsonStore + +__all__ = [ + "Action", + "ActionContext", + "ActionRegistrationError", + "JsonStore", + "RoutineEngine", + "RoutineEngineError", + "RunResult", + "RunStatus", + "Step", + "StepResult", + "StepStatus", + "StorageError", + "Workflow", + "WorkflowValidationError", +] + +__version__ = "0.1.0" diff --git a/src/routine_engine/__main__.py b/src/routine_engine/__main__.py new file mode 100644 index 0000000..3bb3df3 --- /dev/null +++ b/src/routine_engine/__main__.py @@ -0,0 +1,5 @@ +"""Run the CLI with ``python -m routine_engine``.""" + +from .cli import main + +raise SystemExit(main()) diff --git a/src/routine_engine/builtins.py b/src/routine_engine/builtins.py new file mode 100644 index 0000000..3a73b3b --- /dev/null +++ b/src/routine_engine/builtins.py @@ -0,0 +1,40 @@ +"""Small, side-effect-free actions used by the CLI and examples.""" + +from __future__ import annotations + +from typing import Any + +from .engine import RoutineEngine +from .models import ActionContext + + +def identity(context: ActionContext) -> Any: + """Return ``with.value`` or the complete workflow input.""" + + return context.params.get("value", dict(context.inputs)) + + +def merge(context: ActionContext) -> dict[str, Any]: + """Return resolved step parameters as a new mapping.""" + + return dict(context.params) + + +def format_text(context: ActionContext) -> str: + """Format ``with.template`` using the mapping in ``with.values``.""" + + template = context.params.get("template") + values = context.params.get("values", {}) + if not isinstance(template, str): + raise ValueError("format requires a string 'template'") + if not isinstance(values, dict): + raise ValueError("format requires an object named 'values'") + return template.format_map(values) + + +def register_builtin_actions(engine: RoutineEngine) -> None: + """Register the intentionally small, safe built-in action set.""" + + engine.register("identity", identity) + engine.register("merge", merge) + engine.register("format", format_text) diff --git a/src/routine_engine/cli.py b/src/routine_engine/cli.py new file mode 100644 index 0000000..04599ab --- /dev/null +++ b/src/routine_engine/cli.py @@ -0,0 +1,130 @@ +"""Command-line interface for validation and local execution.""" + +from __future__ import annotations + +import argparse +import importlib +import json +import sys +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +from typing import Any, Sequence + +from .builtins import register_builtin_actions +from .engine import RoutineEngine +from .errors import RoutineEngineError, WorkflowValidationError +from .models import RunStatus +from .storage import JsonStore + + +def _version() -> str: + try: + return version("samsarix-routine-engine") + except PackageNotFoundError: + return "0.1.0" + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="routine-engine", + description="Validate and run small, local-first workflow DAGs.", + ) + parser.add_argument("--version", action="version", version=f"%(prog)s {_version()}") + subparsers = parser.add_subparsers(dest="command", required=True) + + validate = subparsers.add_parser("validate", help="validate a workflow JSON file") + validate.add_argument("workflow", type=Path) + validate.add_argument("--plugin", action="append", default=[], metavar="MODULE") + + run = subparsers.add_parser("run", help="run a workflow JSON file") + run.add_argument("workflow", type=Path) + run.add_argument("--input", default="{}", metavar="JSON_OR_@FILE") + run.add_argument("--state", type=Path, help="persist definitions and the last 100 runs") + run.add_argument("--plugin", action="append", default=[], metavar="MODULE") + + demo = subparsers.add_parser("demo", help="run the built-in greeting workflow") + demo.add_argument("--name", default="world") + return parser + + +def _load_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except OSError as exc: + raise WorkflowValidationError(f"cannot read '{path}': {exc}") from exc + except json.JSONDecodeError as exc: + raise WorkflowValidationError(f"'{path}' is not valid JSON: {exc}") from exc + if not isinstance(value, dict): + raise WorkflowValidationError(f"'{path}' must contain a JSON object") + return value + + +def _load_input(raw: str) -> dict[str, Any]: + if raw.startswith("@"): + return _load_json(Path(raw[1:])) + try: + value = json.loads(raw) + except json.JSONDecodeError as exc: + raise WorkflowValidationError(f"--input is not valid JSON: {exc}") from exc + if not isinstance(value, dict): + raise WorkflowValidationError("--input must be a JSON object") + return value + + +def _make_engine(plugins: list[str], state: Path | None = None) -> RoutineEngine: + engine = RoutineEngine(store=JsonStore(state) if state else None) + register_builtin_actions(engine) + for module_name in plugins: + try: + module = importlib.import_module(module_name) + except ImportError as exc: + raise WorkflowValidationError(f"cannot import plugin '{module_name}': {exc}") from exc + register = getattr(module, "register", None) + if not callable(register): + raise WorkflowValidationError(f"plugin '{module_name}' must define register(engine)") + register(engine) + return engine + + +def _demo() -> dict[str, Any]: + return { + "id": "greeting", + "steps": [ + {"id": "name", "action": "identity", "with": {"value": "{{ input.name }}"}}, + { + "id": "greet", + "action": "format", + "needs": ["name"], + "with": { + "template": "Hello, {name}!", + "values": {"name": "{{ steps.name.output }}"}, + }, + }, + ], + } + + +def main(argv: Sequence[str] | None = None) -> int: + args = _parser().parse_args(argv) + try: + if args.command == "demo": + result = _make_engine([]).run(_demo(), {"name": args.name}) + print(json.dumps(result.to_dict(), indent=2)) + return 0 + + workflow = _load_json(args.workflow) + if args.command == "validate": + definition = _make_engine(args.plugin).validate(workflow) + print(f"valid: {definition.id} ({len(definition.steps)} steps)") + return 0 + + result = _make_engine(args.plugin, args.state).run(workflow, _load_input(args.input)) + print(json.dumps(result.to_dict(), indent=2, allow_nan=False)) + return 0 if result.status is RunStatus.SUCCESS else 1 + except (RoutineEngineError, OSError, ValueError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/routine_engine/engine.py b/src/routine_engine/engine.py new file mode 100644 index 0000000..a864a30 --- /dev/null +++ b/src/routine_engine/engine.py @@ -0,0 +1,275 @@ +"""Bounded, dependency-aware workflow execution.""" + +from __future__ import annotations + +import asyncio +import inspect +import json +import re +import uuid +from collections.abc import Awaitable, Callable, Mapping +from types import MappingProxyType +from typing import Any + +from .errors import ActionRegistrationError, WorkflowValidationError +from .models import ( + ActionContext, + RunResult, + RunStatus, + Step, + StepResult, + StepStatus, + Workflow, + utc_now, +) +from .storage import JsonStore + +Action = Callable[[ActionContext], Any | Awaitable[Any]] +_REFERENCE = re.compile(r"^\{\{\s*(input|steps)\.([A-Za-z0-9_.-]+)\s*\}\}$") + + +class RoutineEngine: + """Register trusted Python actions and execute validated workflow DAGs.""" + + def __init__(self, *, store: JsonStore | None = None) -> None: + self._actions: dict[str, Action] = {} + self._store = store + + @property + def actions(self) -> tuple[str, ...]: + return tuple(sorted(self._actions)) + + def register(self, name: str, action: Action, *, replace: bool = False) -> None: + """Register trusted application code under a workflow-visible name.""" + + if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,127}", name): + raise ActionRegistrationError("action name is not a valid identifier") + if not callable(action): + raise ActionRegistrationError(f"action '{name}' must be callable") + if name in self._actions and not replace: + raise ActionRegistrationError(f"action '{name}' is already registered") + self._actions[name] = action + + def validate(self, workflow: Workflow | Mapping[str, Any]) -> Workflow: + """Validate structure, dependency graph, and action registrations.""" + + definition = workflow if isinstance(workflow, Workflow) else Workflow.from_dict(workflow) + missing = sorted({step.action for step in definition.steps} - self._actions.keys()) + if missing: + raise WorkflowValidationError(f"unregistered action(s): {', '.join(missing)}") + return definition + + def run(self, workflow: Workflow | Mapping[str, Any], inputs: Mapping[str, Any] | None = None) -> RunResult: + """Run a workflow from synchronous code.""" + + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(self.arun(workflow, inputs)) + raise RuntimeError("RoutineEngine.run() cannot be called inside an event loop; use 'await arun(...)'") + + async def arun( + self, + workflow: Workflow | Mapping[str, Any], + inputs: Mapping[str, Any] | None = None, + ) -> RunResult: + """Run a workflow and return exact per-step terminal states.""" + + definition = self.validate(workflow) + try: + run_inputs = json.loads(json.dumps(dict(inputs or {}), allow_nan=False)) + except (TypeError, ValueError, OverflowError) as exc: + raise WorkflowValidationError("run inputs must be finite, JSON-compatible data") from exc + run_id = uuid.uuid4().hex + started_at = utc_now() + pending = {step.id: step for step in definition.steps} + results: dict[str, StepResult] = {} + running: dict[str, asyncio.Task[StepResult]] = {} + + if self._store is not None: + self._store.save_workflow(definition) + + try: + while pending or running: + for step_id, step in tuple(pending.items()): + dependency_results = [results.get(dependency) for dependency in step.needs] + if any( + result is not None and result.status is not StepStatus.SUCCESS + for result in dependency_results + ): + now = utc_now() + results[step_id] = StepResult( + step_id=step_id, + status=StepStatus.SKIPPED, + attempts=0, + started_at=now, + finished_at=now, + error="dependency did not succeed", + ) + del pending[step_id] + + capacity = definition.max_concurrency - len(running) + ready = [ + step + for step in definition.steps + if step.id in pending + and all( + dependency in results and results[dependency].status is StepStatus.SUCCESS + for dependency in step.needs + ) + ] + for step in ready[:capacity]: + running[step.id] = asyncio.create_task( + self._run_step(definition, step, run_id, run_inputs, results), + name=f"routine-engine:{run_id}:{step.id}", + ) + del pending[step.id] + + if not running: + if pending: + raise RuntimeError("validated workflow became unschedulable") + break + + done, _ = await asyncio.wait(running.values(), return_when=asyncio.FIRST_COMPLETED) + for task in done: + step_result = task.result() + results[step_result.step_id] = step_result + del running[step_result.step_id] + except asyncio.CancelledError: + for task in running.values(): + task.cancel() + await asyncio.gather(*running.values(), return_exceptions=True) + now = utc_now() + for step_id in (*running.keys(), *pending.keys()): + results[step_id] = StepResult( + step_id=step_id, + status=StepStatus.CANCELLED, + attempts=0, + started_at=now, + finished_at=now, + error="workflow run cancelled", + ) + cancelled = RunResult( + run_id=run_id, + workflow_id=definition.id, + status=RunStatus.CANCELLED, + started_at=started_at, + finished_at=utc_now(), + steps=MappingProxyType(_ordered_results(definition, results)), + ) + if self._store is not None: + self._store.append_run(cancelled) + raise + + status = RunStatus.FAILED if any(r.status is StepStatus.FAILED for r in results.values()) else RunStatus.SUCCESS + result = RunResult( + run_id=run_id, + workflow_id=definition.id, + status=status, + started_at=started_at, + finished_at=utc_now(), + steps=MappingProxyType(_ordered_results(definition, results)), + ) + if self._store is not None: + self._store.append_run(result) + return result + + async def _run_step( + self, + workflow: Workflow, + step: Step, + run_id: str, + inputs: Mapping[str, Any], + results: Mapping[str, StepResult], + ) -> StepResult: + started_at = utc_now() + output_values = {dependency: results[dependency].output for dependency in step.needs} + try: + params = _resolve_value(dict(step.params), inputs, output_values) + except (KeyError, TypeError, WorkflowValidationError) as exc: + return StepResult( + step_id=step.id, + status=StepStatus.FAILED, + attempts=0, + started_at=started_at, + finished_at=utc_now(), + error=f"ParameterResolutionError: {exc}", + ) + + action = self._actions[step.action] + last_error: str | None = None + for attempt in range(1, step.retries + 2): + context = ActionContext( + run_id=run_id, + workflow_id=workflow.id, + step_id=step.id, + attempt=attempt, + inputs=MappingProxyType(dict(inputs)), + params=MappingProxyType(params), + outputs=MappingProxyType(output_values), + ) + try: + value = action(context) + output = await value if inspect.isawaitable(value) else value + return StepResult( + step_id=step.id, + status=StepStatus.SUCCESS, + attempts=attempt, + started_at=started_at, + finished_at=utc_now(), + output=output, + ) + except Exception as exc: # actions define their own expected exception types + last_error = f"{type(exc).__name__}: {exc}" + if attempt <= step.retries and step.retry_delay_seconds: + await asyncio.sleep(step.retry_delay_seconds) + + return StepResult( + step_id=step.id, + status=StepStatus.FAILED, + attempts=step.retries + 1, + started_at=started_at, + finished_at=utc_now(), + error=last_error, + ) + + +def _ordered_results(workflow: Workflow, results: Mapping[str, StepResult]) -> dict[str, StepResult]: + return {step.id: results[step.id] for step in workflow.steps} + + +def _resolve_value(value: Any, inputs: Mapping[str, Any], outputs: Mapping[str, Any]) -> Any: + if isinstance(value, str): + match = _REFERENCE.fullmatch(value) + if match is None: + return value + scope, path = match.groups() + if scope == "input": + return _lookup(inputs, path) + matches = [ + step_id + for step_id in outputs + if path == f"{step_id}.output" or path.startswith(f"{step_id}.output.") + ] + if not matches: + raise WorkflowValidationError(f"reference '{value}' does not name an available step output") + step_id = max(matches, key=len) + output_path = path.removeprefix(f"{step_id}.output").removeprefix(".") + output = outputs[step_id] + return _lookup(output, output_path) if output_path else output + if isinstance(value, list): + return [_resolve_value(item, inputs, outputs) for item in value] + if isinstance(value, dict): + return {key: _resolve_value(item, inputs, outputs) for key, item in value.items()} + return value + + +def _lookup(value: Any, path: str) -> Any: + if not path: + return value + current = value + for segment in path.split("."): + if not isinstance(current, Mapping) or segment not in current: + raise KeyError(f"missing reference segment '{segment}'") + current = current[segment] + return current diff --git a/src/routine_engine/errors.py b/src/routine_engine/errors.py new file mode 100644 index 0000000..d520fa8 --- /dev/null +++ b/src/routine_engine/errors.py @@ -0,0 +1,17 @@ +"""Public exception hierarchy for Routine Engine.""" + + +class RoutineEngineError(Exception): + """Base exception for expected Routine Engine failures.""" + + +class WorkflowValidationError(RoutineEngineError, ValueError): + """Raised when a workflow definition is malformed or unsafe to execute.""" + + +class ActionRegistrationError(RoutineEngineError, ValueError): + """Raised when an action cannot be registered.""" + + +class StorageError(RoutineEngineError): + """Raised when persisted state cannot be read or written safely.""" diff --git a/src/routine_engine/models.py b/src/routine_engine/models.py new file mode 100644 index 0000000..97d6c7e --- /dev/null +++ b/src/routine_engine/models.py @@ -0,0 +1,293 @@ +"""Validated workflow definitions and immutable execution results.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from types import MappingProxyType +from typing import Any, Mapping + +from .errors import WorkflowValidationError + +MAX_STEPS = 256 +MAX_RETRIES = 10 +MAX_RETRY_DELAY_SECONDS = 60.0 +MAX_CONCURRENCY = 32 +_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,127}$") +_INPUT_REFERENCE = re.compile( + r"^\{\{\s*input\.[A-Za-z][A-Za-z0-9_-]*(?:\.[A-Za-z][A-Za-z0-9_-]*)*\s*\}\}$" +) + + +def utc_now() -> datetime: + """Return a timezone-aware UTC timestamp.""" + + return datetime.now(timezone.utc) + + +def _require_identifier(value: Any, field_name: str) -> str: + if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value): + raise WorkflowValidationError( + f"{field_name} must start with a letter and contain only letters, numbers, '.', '_' or '-'" + ) + return value + + +def _require_json(value: Any, field_name: str) -> None: + try: + json.dumps(value, allow_nan=False) + except (TypeError, ValueError, OverflowError) as exc: + raise WorkflowValidationError(f"{field_name} must be finite, JSON-compatible data") from exc + + +class StepStatus(str, Enum): + """Terminal state of one workflow step.""" + + SUCCESS = "success" + FAILED = "failed" + SKIPPED = "skipped" + CANCELLED = "cancelled" + + +class RunStatus(str, Enum): + """Terminal state of a workflow run.""" + + SUCCESS = "success" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass(frozen=True) +class Step: + """A single action invocation in a workflow DAG.""" + + id: str + action: str + needs: tuple[str, ...] = () + params: Mapping[str, Any] = field(default_factory=dict) + retries: int = 0 + retry_delay_seconds: float = 0.0 + + @classmethod + def from_dict(cls, value: Mapping[str, Any], index: int) -> Step: + if not isinstance(value, Mapping): + raise WorkflowValidationError(f"steps[{index}] must be an object") + + step_id = _require_identifier(value.get("id"), f"steps[{index}].id") + action = _require_identifier(value.get("action"), f"steps[{index}].action") + raw_needs = value.get("needs", []) + if not isinstance(raw_needs, list) or not all(isinstance(item, str) for item in raw_needs): + raise WorkflowValidationError(f"steps[{index}].needs must be an array of step IDs") + if len(raw_needs) != len(set(raw_needs)): + raise WorkflowValidationError(f"steps[{index}].needs contains duplicates") + + params = value.get("with", {}) + if not isinstance(params, Mapping): + raise WorkflowValidationError(f"steps[{index}].with must be an object") + _require_json(params, f"steps[{index}].with") + + retries = value.get("retries", 0) + if isinstance(retries, bool) or not isinstance(retries, int) or not 0 <= retries <= MAX_RETRIES: + raise WorkflowValidationError(f"steps[{index}].retries must be an integer from 0 to {MAX_RETRIES}") + + delay = value.get("retry_delay_seconds", 0.0) + if isinstance(delay, bool) or not isinstance(delay, (int, float)): + raise WorkflowValidationError(f"steps[{index}].retry_delay_seconds must be a number") + delay = float(delay) + if not 0.0 <= delay <= MAX_RETRY_DELAY_SECONDS: + raise WorkflowValidationError( + f"steps[{index}].retry_delay_seconds must be from 0 to {MAX_RETRY_DELAY_SECONDS:g}" + ) + + detached_params = json.loads(json.dumps(params, allow_nan=False)) + return cls( + id=step_id, + action=action, + needs=tuple(raw_needs), + params=MappingProxyType(detached_params), + retries=retries, + retry_delay_seconds=delay, + ) + + def to_dict(self) -> dict[str, Any]: + value: dict[str, Any] = {"id": self.id, "action": self.action} + if self.needs: + value["needs"] = list(self.needs) + if self.params: + value["with"] = dict(self.params) + if self.retries: + value["retries"] = self.retries + if self.retry_delay_seconds: + value["retry_delay_seconds"] = self.retry_delay_seconds + return value + + +@dataclass(frozen=True) +class Workflow: + """A validated, acyclic workflow definition.""" + + id: str + steps: tuple[Step, ...] + description: str = "" + max_concurrency: int = 4 + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> Workflow: + if not isinstance(value, Mapping): + raise WorkflowValidationError("workflow must be a JSON object") + workflow_id = _require_identifier(value.get("id"), "id") + description = value.get("description", "") + if not isinstance(description, str) or len(description) > 1000: + raise WorkflowValidationError("description must be a string of at most 1000 characters") + + raw_steps = value.get("steps") + if not isinstance(raw_steps, list) or not raw_steps: + raise WorkflowValidationError("steps must be a non-empty array") + if len(raw_steps) > MAX_STEPS: + raise WorkflowValidationError(f"a workflow may contain at most {MAX_STEPS} steps") + steps = tuple(Step.from_dict(item, index) for index, item in enumerate(raw_steps)) + + ids = {step.id for step in steps} + if len(ids) != len(steps): + raise WorkflowValidationError("step IDs must be unique") + for step in steps: + if step.id in step.needs: + raise WorkflowValidationError(f"step '{step.id}' cannot depend on itself") + unknown = set(step.needs) - ids + if unknown: + raise WorkflowValidationError( + f"step '{step.id}' depends on unknown step(s): {', '.join(sorted(unknown))}" + ) + referenced = _referenced_steps(step.params, ids) + undeclared = referenced - set(step.needs) + if undeclared: + raise WorkflowValidationError( + f"step '{step.id}' must list referenced step(s) in needs: {', '.join(sorted(undeclared))}" + ) + _validate_acyclic(steps) + + max_concurrency = value.get("max_concurrency", 4) + if ( + isinstance(max_concurrency, bool) + or not isinstance(max_concurrency, int) + or not 1 <= max_concurrency <= MAX_CONCURRENCY + ): + raise WorkflowValidationError(f"max_concurrency must be an integer from 1 to {MAX_CONCURRENCY}") + + return cls( + id=workflow_id, + steps=steps, + description=description, + max_concurrency=max_concurrency, + ) + + def to_dict(self) -> dict[str, Any]: + value: dict[str, Any] = {"id": self.id, "steps": [step.to_dict() for step in self.steps]} + if self.description: + value["description"] = self.description + if self.max_concurrency != 4: + value["max_concurrency"] = self.max_concurrency + return value + + +def _validate_acyclic(steps: tuple[Step, ...]) -> None: + dependencies = {step.id: set(step.needs) for step in steps} + complete: set[str] = set() + while len(complete) < len(steps): + ready = {step_id for step_id, needs in dependencies.items() if step_id not in complete and needs <= complete} + if not ready: + blocked = ", ".join(sorted(set(dependencies) - complete)) + raise WorkflowValidationError(f"workflow contains a dependency cycle involving: {blocked}") + complete.update(ready) + + +def _referenced_steps(value: Any, step_ids: set[str]) -> set[str]: + if isinstance(value, str): + stripped = value.strip() + if "{{" not in stripped and "}}" not in stripped: + return set() + if _INPUT_REFERENCE.fullmatch(stripped): + return set() + if stripped.startswith("{{") and stripped.endswith("}}"): + expression = stripped[2:-2].strip() + if expression.startswith("steps."): + path = expression.removeprefix("steps.") + matches = { + step_id + for step_id in step_ids + if path == f"{step_id}.output" or path.startswith(f"{step_id}.output.") + } + if matches: + return {max(matches, key=len)} + raise WorkflowValidationError(f"invalid reference syntax: '{value}'") + if isinstance(value, list): + return set().union(*(_referenced_steps(item, step_ids) for item in value), set()) + if isinstance(value, Mapping): + return set().union(*(_referenced_steps(item, step_ids) for item in value.values()), set()) + return set() + + +@dataclass(frozen=True) +class ActionContext: + """Inputs available to a registered action.""" + + run_id: str + workflow_id: str + step_id: str + attempt: int + inputs: Mapping[str, Any] + params: Mapping[str, Any] + outputs: Mapping[str, Any] + + +@dataclass(frozen=True) +class StepResult: + """The auditable result of one step.""" + + step_id: str + status: StepStatus + attempts: int + started_at: datetime + finished_at: datetime + output: Any = None + error: str | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "step_id": self.step_id, + "status": self.status.value, + "attempts": self.attempts, + "started_at": self.started_at.isoformat(), + "finished_at": self.finished_at.isoformat(), + "output": self.output, + "error": self.error, + } + + +@dataclass(frozen=True) +class RunResult: + """The terminal result of a complete workflow run.""" + + run_id: str + workflow_id: str + status: RunStatus + started_at: datetime + finished_at: datetime + steps: Mapping[str, StepResult] + + @property + def succeeded(self) -> bool: + return self.status is RunStatus.SUCCESS + + def to_dict(self) -> dict[str, Any]: + return { + "run_id": self.run_id, + "workflow_id": self.workflow_id, + "status": self.status.value, + "started_at": self.started_at.isoformat(), + "finished_at": self.finished_at.isoformat(), + "steps": {step_id: result.to_dict() for step_id, result in self.steps.items()}, + } diff --git a/src/routine_engine/py.typed b/src/routine_engine/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/routine_engine/py.typed @@ -0,0 +1 @@ + diff --git a/src/routine_engine/storage.py b/src/routine_engine/storage.py new file mode 100644 index 0000000..6d1264c --- /dev/null +++ b/src/routine_engine/storage.py @@ -0,0 +1,95 @@ +"""Atomic local JSON persistence for workflows and bounded run history.""" + +from __future__ import annotations + +import json +import os +import tempfile +import threading +from pathlib import Path +from typing import Any, cast + +from .errors import StorageError +from .models import RunResult, Workflow + + +class JsonStore: + """Persist definitions and recent results in one human-readable JSON file. + + Writes are atomic within one filesystem. A store instance is thread-safe, but + concurrent writers in separate processes require external coordination. + """ + + def __init__(self, path: str | Path, *, history_limit: int = 100) -> None: + if not 1 <= history_limit <= 10_000: + raise ValueError("history_limit must be from 1 to 10000") + self.path = Path(path).expanduser().resolve() + self.history_limit = history_limit + self._lock = threading.RLock() + + def save_workflow(self, workflow: Workflow) -> None: + with self._lock: + state = self._read() + state["workflows"][workflow.id] = workflow.to_dict() + self._write(state) + + def append_run(self, run: RunResult) -> None: + with self._lock: + state = self._read() + state["runs"].append(run.to_dict()) + state["runs"] = state["runs"][-self.history_limit :] + self._write(state) + + def snapshot(self) -> dict[str, Any]: + """Return a detached copy of current persisted state.""" + + with self._lock: + return cast(dict[str, Any], json.loads(json.dumps(self._read(), allow_nan=False))) + + def _read(self) -> dict[str, Any]: + if not self.path.exists(): + return {"schema_version": 1, "workflows": {}, "runs": []} + try: + value = json.loads(self.path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise StorageError(f"cannot read state from '{self.path}': {exc}") from exc + if ( + not isinstance(value, dict) + or value.get("schema_version") != 1 + or not isinstance(value.get("workflows"), dict) + or not isinstance(value.get("runs"), list) + ): + raise StorageError(f"state file '{self.path}' does not match schema version 1") + return value + + def _write(self, state: dict[str, Any]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + try: + payload = json.dumps(state, indent=2, sort_keys=True, allow_nan=False) + "\n" + except (TypeError, ValueError, OverflowError) as exc: + raise StorageError("workflow outputs must be finite, JSON-compatible data to persist them") from exc + + temp_path: str | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + newline="\n", + dir=self.path.parent, + prefix=f".{self.path.name}.", + suffix=".tmp", + delete=False, + ) as handle: + temp_path = handle.name + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_path, self.path) + except OSError as exc: + raise StorageError(f"cannot write state to '{self.path}': {exc}") from exc + finally: + if temp_path is not None: + try: + Path(temp_path).unlink(missing_ok=True) + except OSError: + pass diff --git a/tests/conftest.py b/tests/conftest.py index 84d4567..1df6017 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,209 +1,33 @@ -"""Comprehensive pytest configuration and fixtures for routine-engine.""" +from __future__ import annotations -import pytest -from unittest.mock import Mock, MagicMock, patch -from datetime import datetime, timedelta - - -# ============================================================================ -# Workflow Fixtures -# ============================================================================ - -@pytest.fixture -def mock_workflow(): - """Mock workflow.""" - workflow = MagicMock() - workflow.id = "workflow-1" - workflow.name = "TestWorkflow" - workflow.status = "active" - workflow.execute = MagicMock(return_value={"result": "success"}) - workflow.get_status = MagicMock(return_value="active") - return workflow - - -@pytest.fixture -def mock_workflow_config(): - """Mock workflow configuration.""" - return { - "name": "TestWorkflow", - "steps": [ - {"id": "step-1", "type": "agent", "action": "analyze"}, - {"id": "step-2", "type": "agent", "action": "plan"} - ], - "schedule": "0 0 * * *", - "enabled": True - } - - -# ============================================================================ -# Task Fixtures -# ============================================================================ - -@pytest.fixture -def mock_task(): - """Mock task.""" - task = MagicMock() - task.id = "task-1" - task.name = "TestTask" - task.status = "pending" - task.execute = MagicMock(return_value={"result": "success"}) - task.get_result = MagicMock(return_value={"output": "data"}) - return task - - -@pytest.fixture -def mock_task_config(): - """Mock task configuration.""" - return { - "name": "TestTask", - "action": "execute_analysis", - "parameters": {"input": "data"}, - "timeout": 300, - "retries": 3 - } - - -# ============================================================================ -# Schedule Fixtures -# ============================================================================ - -@pytest.fixture -def mock_schedule(): - """Mock schedule.""" - schedule = MagicMock() - schedule.id = "schedule-1" - schedule.cron = "0 0 * * *" - schedule.enabled = True - schedule.next_run = datetime.now() + timedelta(hours=1) - return schedule - - -@pytest.fixture -def mock_schedule_config(): - """Mock schedule configuration.""" - return { - "cron": "0 0 * * *", - "timezone": "UTC", - "enabled": True, - "max_instances": 1 - } +from collections.abc import Callable +from typing import Any +import pytest -# ============================================================================ -# Agent Fixtures -# ============================================================================ - -@pytest.fixture -def mock_agent(): - """Mock agent.""" - agent = MagicMock() - agent.id = "agent-1" - agent.name = "TestAgent" - agent.execute = MagicMock(return_value={"result": "success"}) - agent.get_state = MagicMock(return_value={"status": "ready"}) - return agent - - -@pytest.fixture -def mock_agents_list(): - """Mock list of agents.""" - agents = [] - for i in range(3): - agent = MagicMock() - agent.id = f"agent-{i}" - agent.name = f"Agent{i}" - agents.append(agent) - return agents - - -# ============================================================================ -# Engine Fixtures -# ============================================================================ - -@pytest.fixture -def mock_workflow_engine(): - """Mock workflow engine.""" - engine = MagicMock() - engine.create_workflow = MagicMock(return_value="workflow-1") - engine.execute_workflow = MagicMock(return_value={"result": "success"}) - engine.get_workflow = MagicMock(return_value=MagicMock()) - engine.list_workflows = MagicMock(return_value=[]) - engine.delete_workflow = MagicMock(return_value=True) - return engine - - -@pytest.fixture -def mock_scheduler(): - """Mock scheduler.""" - scheduler = MagicMock() - scheduler.schedule_workflow = MagicMock(return_value="schedule-1") - scheduler.get_schedule = MagicMock(return_value=MagicMock()) - scheduler.list_schedules = MagicMock(return_value=[]) - scheduler.cancel_schedule = MagicMock(return_value=True) - return scheduler - - -# ============================================================================ -# Execution Fixtures -# ============================================================================ - -@pytest.fixture -def mock_execution_result(): - """Mock execution result.""" - return { - "id": "exec-1", - "workflow_id": "workflow-1", - "status": "success", - "result": {"output": "data"}, - "duration": 1.5, - "timestamp": datetime.now().isoformat() - } +from routine_engine import ActionContext, RoutineEngine @pytest.fixture -def mock_execution_history(): - """Mock execution history.""" - return [ - {"id": "exec-1", "status": "success", "duration": 1.5}, - {"id": "exec-2", "status": "success", "duration": 2.0}, - {"id": "exec-3", "status": "failed", "duration": 0.5} - ] - +def engine() -> RoutineEngine: + instance = RoutineEngine() + instance.register("echo", lambda context: context.params.get("value", dict(context.inputs))) + instance.register("merge", lambda context: dict(context.params)) + return instance -# ============================================================================ -# Scenario Fixtures -# ============================================================================ @pytest.fixture -def workflow_execution_scenario(): - """Workflow execution scenario.""" - return { - "workflow_id": "workflow-1", - "steps": 5, - "expected_duration": 10.0, - "timeout": 30, - "should_succeed": True - } +def workflow_factory() -> Callable[..., dict[str, Any]]: + def make(**overrides: Any) -> dict[str, Any]: + workflow: dict[str, Any] = { + "id": "example", + "steps": [{"id": "first", "action": "echo"}], + } + workflow.update(overrides) + return workflow + return make -@pytest.fixture -def scheduled_workflow_scenario(): - """Scheduled workflow scenario.""" - return { - "workflow_id": "workflow-1", - "schedule": "0 0 * * *", - "timezone": "UTC", - "next_run": datetime.now() + timedelta(hours=1), - "enabled": True - } - -@pytest.fixture -def error_recovery_scenario(): - """Error recovery scenario.""" - return { - "error_type": "timeout", - "retry_count": 3, - "retry_delay": 5, - "should_recover": True - } +def output(context: ActionContext) -> Any: + return context.params.get("value") diff --git a/tests/test_execution.py b/tests/test_execution.py index a4cc228..14dc9ef 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -1,44 +1,223 @@ -"""Test suite for execution functionality.""" +from __future__ import annotations + +import asyncio import pytest +from routine_engine import ( + ActionContext, + ActionRegistrationError, + RoutineEngine, + RunStatus, + StepStatus, + WorkflowValidationError, +) + + +def test_primary_journey_resolves_input_and_prior_outputs(engine: RoutineEngine) -> None: + workflow = { + "id": "welcome", + "steps": [ + {"id": "name", "action": "echo", "with": {"value": "{{ input.user.name }}"}}, + { + "id": "result", + "action": "merge", + "needs": ["name"], + "with": {"message": "{{ steps.name.output }}", "source": "local"}, + }, + ], + } + + result = engine.run(workflow, {"user": {"name": "Ada"}}) + + assert result.status is RunStatus.SUCCESS + assert result.succeeded + assert result.steps["name"].output == "Ada" + assert result.steps["result"].output == {"message": "Ada", "source": "local"} + assert list(result.steps) == ["name", "result"] + assert result.to_dict()["status"] == "success" + + +def test_dotted_step_id_resolves_without_ambiguity(engine: RoutineEngine) -> None: + result = engine.run( + { + "id": "dotted", + "steps": [ + {"id": "source.v1", "action": "echo", "with": {"value": {"name": "Ada"}}}, + { + "id": "consumer", + "action": "echo", + "needs": ["source.v1"], + "with": {"value": "{{ steps.source.v1.output.name }}"}, + }, + ], + } + ) + + assert result.steps["consumer"].output == "Ada" + + +def test_inputs_must_be_json_and_are_detached(engine: RoutineEngine) -> None: + inputs = {"nested": {"value": "original"}} + + def mutate(context: ActionContext) -> str: + nested = context.inputs["nested"] + assert isinstance(nested, dict) + nested["value"] = "changed" + return nested["value"] + + engine.register("mutate", mutate) + result = engine.run({"id": "detach", "steps": [{"id": "x", "action": "mutate"}]}, inputs) + assert result.succeeded + assert inputs == {"nested": {"value": "original"}} + + with pytest.raises(WorkflowValidationError, match="JSON-compatible"): + engine.run({"id": "bad-input", "steps": [{"id": "x", "action": "echo"}]}, {"bad": {1}}) + + +def test_action_context_exposes_only_declared_dependency_outputs() -> None: + engine = RoutineEngine() + engine.register("value", lambda context: context.step_id) + engine.register("keys", lambda context: sorted(context.outputs)) + + result = engine.run( + { + "id": "outputs", + "max_concurrency": 1, + "steps": [ + {"id": "unrelated", "action": "value"}, + {"id": "source", "action": "value"}, + {"id": "consumer", "action": "keys", "needs": ["source"]}, + ], + } + ) + + assert result.steps["consumer"].output == ["source"] + + +@pytest.mark.asyncio +async def test_async_actions_run_in_parallel() -> None: + engine = RoutineEngine() + both_started = asyncio.Event() + starts = 0 + + async def rendezvous(context: ActionContext) -> str: + nonlocal starts + starts += 1 + if starts == 2: + both_started.set() + await asyncio.wait_for(both_started.wait(), timeout=0.5) + return context.step_id + + engine.register("rendezvous", rendezvous) + result = await engine.arun( + { + "id": "parallel", + "max_concurrency": 2, + "steps": [ + {"id": "left", "action": "rendezvous"}, + {"id": "right", "action": "rendezvous"}, + ], + } + ) + + assert result.status is RunStatus.SUCCESS + assert {item.output for item in result.steps.values()} == {"left", "right"} + + +@pytest.mark.asyncio +async def test_retry_then_success_records_attempt_count() -> None: + engine = RoutineEngine() + + def flaky(context: ActionContext) -> int: + if context.attempt < 3: + raise RuntimeError("not yet") + return context.attempt + + engine.register("flaky", flaky) + result = await engine.arun( + { + "id": "retry", + "steps": [{"id": "eventual", "action": "flaky", "retries": 2, "retry_delay_seconds": 0.001}], + } + ) + + assert result.steps["eventual"].status is StepStatus.SUCCESS + assert result.steps["eventual"].attempts == 3 + + +def test_failure_skips_dependents_but_independent_branch_completes() -> None: + engine = RoutineEngine() + + def fail(_: ActionContext) -> None: + raise LookupError("boom") + + engine.register("fail", fail) + engine.register("ok", lambda _: "done") + result = engine.run( + { + "id": "failure", + "steps": [ + {"id": "bad", "action": "fail", "retries": 1}, + {"id": "blocked", "action": "ok", "needs": ["bad"]}, + {"id": "independent", "action": "ok"}, + ], + } + ) + + assert result.status is RunStatus.FAILED + assert not result.succeeded + assert result.steps["bad"].attempts == 2 + assert result.steps["bad"].error == "LookupError: boom" + assert result.steps["blocked"].status is StepStatus.SKIPPED + assert result.steps["independent"].status is StepStatus.SUCCESS + + +def test_bad_reference_is_a_step_failure(engine: RoutineEngine) -> None: + result = engine.run( + { + "id": "reference", + "steps": [{"id": "step", "action": "echo", "with": {"value": "{{ input.missing }}"}}], + } + ) + + assert result.status is RunStatus.FAILED + assert result.steps["step"].attempts == 0 + assert result.steps["step"].error is not None + assert result.steps["step"].error.startswith("ParameterResolutionError") + + +def test_registration_guards_names_duplicates_and_callables() -> None: + engine = RoutineEngine() + engine.register("valid", lambda _: None) + with pytest.raises(ActionRegistrationError, match="already"): + engine.register("valid", lambda _: None) + with pytest.raises(ActionRegistrationError, match="identifier"): + engine.register("bad name", lambda _: None) + with pytest.raises(ActionRegistrationError, match="callable"): + engine.register("other", None) # type: ignore[arg-type] + engine.register("valid", lambda _: "replaced", replace=True) + assert engine.actions == ("valid",) + + +@pytest.mark.asyncio +async def test_sync_entrypoint_rejects_running_event_loop(engine: RoutineEngine) -> None: + with pytest.raises(RuntimeError, match="await arun"): + engine.run({"id": "w", "steps": [{"id": "x", "action": "echo"}]}) + + +@pytest.mark.asyncio +async def test_cancellation_propagates_and_cancels_actions() -> None: + engine = RoutineEngine() + started = asyncio.Event() + + async def wait_forever(_: ActionContext) -> None: + started.set() + await asyncio.Event().wait() -class TestExecution: - """Test execution.""" - - @pytest.mark.execution - def test_execution_result(self, mock_execution_result): - """Test execution result.""" - assert mock_execution_result["status"] == "success" - assert mock_execution_result["result"]["output"] == "data" - - @pytest.mark.execution - def test_execution_duration(self, mock_execution_result): - """Test execution duration.""" - assert mock_execution_result["duration"] > 0 - - -class TestExecutionHistory: - """Test execution history.""" - - @pytest.mark.execution - def test_execution_history(self, mock_execution_history): - """Test execution history.""" - assert len(mock_execution_history) == 3 - assert mock_execution_history[0]["status"] == "success" - - @pytest.mark.execution - def test_execution_failure(self, mock_execution_history): - """Test execution failure.""" - failed = [e for e in mock_execution_history if e["status"] == "failed"] - assert len(failed) == 1 - - -class TestWorkflowExecution: - """Test workflow execution.""" - - @pytest.mark.integration - def test_workflow_execution_scenario(self, workflow_execution_scenario): - """Test workflow execution scenario.""" - assert workflow_execution_scenario["should_succeed"] is True - assert workflow_execution_scenario["steps"] == 5 + engine.register("wait", wait_forever) + task = asyncio.create_task(engine.arun({"id": "cancel", "steps": [{"id": "x", "action": "wait"}]})) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task diff --git a/tests/test_scheduling.py b/tests/test_scheduling.py index b4673ac..434d95f 100644 --- a/tests/test_scheduling.py +++ b/tests/test_scheduling.py @@ -1,67 +1,128 @@ -"""Test suite for scheduling functionality.""" +from __future__ import annotations + +import json +from pathlib import Path import pytest +from routine_engine import ActionContext, JsonStore, RoutineEngine, StorageError +from routine_engine.cli import main + + +def test_json_store_persists_definition_and_bounded_history(tmp_path: Path) -> None: + path = tmp_path / "nested" / "state.json" + engine = RoutineEngine(store=JsonStore(path, history_limit=2)) + engine.register("ok", lambda _: {"saved": True}) + workflow = {"id": "stored", "steps": [{"id": "one", "action": "ok"}]} + + for _ in range(3): + engine.run(workflow) + + snapshot = JsonStore(path).snapshot() + assert snapshot["schema_version"] == 1 + assert snapshot["workflows"]["stored"] == workflow + assert len(snapshot["runs"]) == 2 + assert all(run["status"] == "success" for run in snapshot["runs"]) + + +def test_json_store_rejects_corrupt_schema_and_non_json_output(tmp_path: Path) -> None: + corrupt = tmp_path / "corrupt.json" + corrupt.write_text("not json", encoding="utf-8") + with pytest.raises(StorageError, match="cannot read"): + JsonStore(corrupt).snapshot() + + wrong = tmp_path / "wrong.json" + wrong.write_text('{"schema_version": 2}', encoding="utf-8") + with pytest.raises(StorageError, match="schema version 1"): + JsonStore(wrong).snapshot() + + engine = RoutineEngine(store=JsonStore(tmp_path / "state.json")) + engine.register("set", lambda _: {1}) + with pytest.raises(StorageError, match="JSON-compatible"): + engine.run({"id": "bad", "steps": [{"id": "one", "action": "set"}]}) + + +def test_store_constructor_bounds_history(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="history_limit"): + JsonStore(tmp_path / "state.json", history_limit=0) + + +def test_cli_validate_run_demo_and_version(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + workflow_path = tmp_path / "workflow.json" + workflow_path.write_text( + json.dumps( + { + "id": "cli", + "steps": [ + {"id": "name", "action": "identity", "with": {"value": "{{ input.name }}"}}, + { + "id": "out", + "action": "merge", + "needs": ["name"], + "with": {"name": "{{ steps.name.output }}"}, + }, + ], + } + ), + encoding="utf-8", + ) + assert main(["validate", str(workflow_path)]) == 0 + assert "valid: cli (2 steps)" in capsys.readouterr().out + + state = tmp_path / "state.json" + assert main(["run", str(workflow_path), "--input", '{"name":"Ada"}', "--state", str(state)]) == 0 + output = json.loads(capsys.readouterr().out) + assert output["steps"]["out"]["output"] == {"name": "Ada"} + assert state.exists() + + assert main(["demo", "--name", "Sam"]) == 0 + assert "Hello, Sam!" in capsys.readouterr().out + + with pytest.raises(SystemExit) as raised: + main(["--version"]) + assert raised.value.code == 0 + + +def test_cli_reports_bad_input_and_bad_plugin(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + workflow_path = tmp_path / "workflow.json" + workflow_path.write_text('{"id":"w","steps":[{"id":"x","action":"identity"}]}', encoding="utf-8") + + assert main(["run", str(workflow_path), "--input", "[]"]) == 2 + assert "JSON object" in capsys.readouterr().err + assert main(["validate", str(workflow_path), "--plugin", "definitely_missing_plugin"]) == 2 + assert "cannot import plugin" in capsys.readouterr().err + + +def test_cli_loads_explicit_plugin(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + plugin = tmp_path / "sample_plugin.py" + plugin.write_text( + "def register(engine):\n engine.register('custom', lambda context: context.inputs['value'])\n", + encoding="utf-8", + ) + workflow = tmp_path / "workflow.json" + workflow.write_text('{"id":"w","steps":[{"id":"x","action":"custom"}]}', encoding="utf-8") + monkeypatch.syspath_prepend(str(tmp_path)) + + assert main(["validate", str(workflow), "--plugin", "sample_plugin"]) == 0 + + +def test_cancelled_run_is_persisted(tmp_path: Path) -> None: + import asyncio + + async def scenario() -> None: + engine = RoutineEngine(store=JsonStore(tmp_path / "state.json")) + started = asyncio.Event() + + async def block(_: ActionContext) -> None: + started.set() + await asyncio.Event().wait() + + engine.register("block", block) + task = asyncio.create_task(engine.arun({"id": "cancel", "steps": [{"id": "x", "action": "block"}]})) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task -class TestScheduleCreation: - """Test schedule creation.""" - - @pytest.mark.schedule - def test_schedule_creation(self, mock_schedule): - """Test schedule creation.""" - assert mock_schedule.id == "schedule-1" - assert mock_schedule.cron == "0 0 * * *" - - @pytest.mark.schedule - def test_schedule_enabled(self, mock_schedule): - """Test schedule enabled status.""" - assert mock_schedule.enabled is True - - -class TestScheduleConfiguration: - """Test schedule configuration.""" - - @pytest.mark.schedule - def test_schedule_config(self, mock_schedule_config): - """Test schedule configuration.""" - assert mock_schedule_config["cron"] == "0 0 * * *" - assert mock_schedule_config["timezone"] == "UTC" - assert mock_schedule_config["enabled"] is True - - -class TestScheduler: - """Test scheduler.""" - - @pytest.mark.schedule - def test_schedule_workflow(self, mock_scheduler): - """Test scheduling workflow.""" - result = mock_scheduler.schedule_workflow() - assert result == "schedule-1" - - @pytest.mark.schedule - def test_get_schedule(self, mock_scheduler): - """Test getting schedule.""" - schedule = mock_scheduler.get_schedule() - assert schedule is not None - - @pytest.mark.schedule - def test_list_schedules(self, mock_scheduler): - """Test listing schedules.""" - schedules = mock_scheduler.list_schedules() - assert isinstance(schedules, list) - - @pytest.mark.schedule - def test_cancel_schedule(self, mock_scheduler): - """Test canceling schedule.""" - result = mock_scheduler.cancel_schedule() - assert result is True - - -class TestScheduledWorkflow: - """Test scheduled workflow.""" - - @pytest.mark.integration - def test_scheduled_workflow_scenario(self, scheduled_workflow_scenario): - """Test scheduled workflow scenario.""" - assert scheduled_workflow_scenario["enabled"] is True - assert scheduled_workflow_scenario["schedule"] == "0 0 * * *" + asyncio.run(scenario()) + assert JsonStore(tmp_path / "state.json").snapshot()["runs"][0]["status"] == "cancelled" diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 08424e8..17e3d25 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -1,45 +1,60 @@ -"""Test suite for task functionality.""" +from __future__ import annotations + +from typing import Any import pytest +from routine_engine import ActionContext, RoutineEngine, RunStatus +from routine_engine.builtins import format_text, identity, merge, register_builtin_actions + + +def _context(params: dict[str, Any], inputs: dict[str, Any] | None = None) -> ActionContext: + return ActionContext( + run_id="run", + workflow_id="workflow", + step_id="step", + attempt=1, + inputs=inputs or {}, + params=params, + outputs={}, + ) + + +def test_builtin_actions_are_small_and_side_effect_free() -> None: + assert identity(_context({"value": 3})) == 3 + assert identity(_context({}, {"name": "Ada"})) == {"name": "Ada"} + assert merge(_context({"a": 1})) == {"a": 1} + assert format_text(_context({"template": "Hi {name}", "values": {"name": "Ada"}})) == "Hi Ada" + + +def test_format_builtin_validates_parameters() -> None: + with pytest.raises(ValueError, match="template"): + format_text(_context({})) + with pytest.raises(ValueError, match="values"): + format_text(_context({"template": "x", "values": []})) + + +def test_builtin_registration_supports_complete_demo_journey() -> None: + engine = RoutineEngine() + register_builtin_actions(engine) + result = engine.run( + { + "id": "demo", + "steps": [ + {"id": "name", "action": "identity", "with": {"value": "{{ input.name }}"}}, + { + "id": "greeting", + "action": "format", + "needs": ["name"], + "with": { + "template": "Hello, {name}!", + "values": {"name": "{{ steps.name.output }}"}, + }, + }, + ], + }, + {"name": "Samsarix"}, + ) -class TestTaskCreation: - """Test task creation.""" - - @pytest.mark.task - def test_task_creation(self, mock_task): - """Test task creation.""" - assert mock_task.id == "task-1" - assert mock_task.name == "TestTask" - - @pytest.mark.task - def test_task_status(self, mock_task): - """Test task status.""" - assert mock_task.status == "pending" - - -class TestTaskExecution: - """Test task execution.""" - - @pytest.mark.task - def test_execute_task(self, mock_task): - """Test task execution.""" - result = mock_task.execute() - assert result["result"] == "success" - - @pytest.mark.task - def test_get_task_result(self, mock_task): - """Test getting task result.""" - result = mock_task.get_result() - assert result["output"] == "data" - - -class TestTaskConfiguration: - """Test task configuration.""" - - @pytest.mark.task - def test_task_config(self, mock_task_config): - """Test task configuration.""" - assert mock_task_config["name"] == "TestTask" - assert mock_task_config["timeout"] == 300 - assert mock_task_config["retries"] == 3 + assert result.status is RunStatus.SUCCESS + assert result.steps["greeting"].output == "Hello, Samsarix!" diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 0dfb48e..f39e847 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -1,73 +1,136 @@ -"""Test suite for workflow functionality.""" +from __future__ import annotations + +from typing import Any import pytest +from routine_engine import RoutineEngine, Workflow, WorkflowValidationError +from routine_engine.models import MAX_CONCURRENCY, MAX_STEPS + + +def test_workflow_round_trip_preserves_public_shape(workflow_factory: Any) -> None: + raw = workflow_factory( + description="A real workflow", + max_concurrency=2, + steps=[ + { + "id": "first", + "action": "echo", + "with": {"value": 3}, + "retries": 2, + "retry_delay_seconds": 0.25, + }, + {"id": "second", "action": "merge", "needs": ["first"]}, + ], + ) + + definition = Workflow.from_dict(raw) + + assert definition.to_dict() == raw + + +@pytest.mark.parametrize( + ("raw", "message"), + [ + ([], "JSON object"), + ({"id": "bad id", "steps": [{}]}, "id must start"), + ({"id": "ok"}, "non-empty array"), + ({"id": "ok", "steps": []}, "non-empty array"), + ({"id": "ok", "steps": [None]}, "must be an object"), + ({"id": "ok", "steps": [{"id": "x", "action": "echo", "needs": "y"}]}, "array"), + ( + {"id": "ok", "steps": [{"id": "x", "action": "echo", "needs": ["y", "y"]}]}, + "duplicates", + ), + ({"id": "ok", "steps": [{"id": "x", "action": "echo", "with": []}]}, "object"), + ({"id": "ok", "steps": [{"id": "x", "action": "echo", "retries": True}]}, "integer"), + ({"id": "ok", "steps": [{"id": "x", "action": "echo", "retries": 11}]}, "from 0"), + ( + {"id": "ok", "steps": [{"id": "x", "action": "echo", "retry_delay_seconds": "1"}]}, + "number", + ), + ( + {"id": "ok", "steps": [{"id": "x", "action": "echo", "retry_delay_seconds": 61}]}, + "must be from", + ), + ( + {"id": "ok", "steps": [{"id": "x", "action": "echo"}], "description": 3}, + "description", + ), + ( + {"id": "ok", "steps": [{"id": "x", "action": "echo"}], "max_concurrency": 0}, + "max_concurrency", + ), + ], +) +def test_rejects_malformed_definitions(raw: Any, message: str) -> None: + with pytest.raises(WorkflowValidationError, match=message): + Workflow.from_dict(raw) + + +def test_rejects_duplicate_unknown_self_and_cyclic_dependencies() -> None: + cases = [ + ( + {"id": "w", "steps": [{"id": "a", "action": "echo"}, {"id": "a", "action": "echo"}]}, + "unique", + ), + ({"id": "w", "steps": [{"id": "a", "action": "echo", "needs": ["missing"]}]}, "unknown"), + ({"id": "w", "steps": [{"id": "a", "action": "echo", "needs": ["a"]}]}, "itself"), + ( + { + "id": "w", + "steps": [ + {"id": "a", "action": "echo", "needs": ["b"]}, + {"id": "b", "action": "echo", "needs": ["a"]}, + ], + }, + "cycle", + ), + ] + for raw, message in cases: + with pytest.raises(WorkflowValidationError, match=message): + Workflow.from_dict(raw) + + +def test_rejects_non_json_params_and_resource_excess() -> None: + with pytest.raises(WorkflowValidationError, match="JSON-compatible"): + Workflow.from_dict({"id": "w", "steps": [{"id": "a", "action": "echo", "with": {"x": {1}}}]}) + + too_many = [{"id": f"s{index}", "action": "echo"} for index in range(MAX_STEPS + 1)] + with pytest.raises(WorkflowValidationError, match=str(MAX_STEPS)): + Workflow.from_dict({"id": "w", "steps": too_many}) + + with pytest.raises(WorkflowValidationError, match=str(MAX_CONCURRENCY)): + Workflow.from_dict( + {"id": "w", "steps": [{"id": "a", "action": "echo"}], "max_concurrency": MAX_CONCURRENCY + 1} + ) + + +def test_engine_requires_registered_actions(engine: RoutineEngine, workflow_factory: Any) -> None: + with pytest.raises(WorkflowValidationError, match=r"unregistered action.*missing"): + engine.validate(workflow_factory(steps=[{"id": "x", "action": "missing"}])) + + +def test_references_require_valid_syntax_and_declared_dependencies() -> None: + with pytest.raises(WorkflowValidationError, match="must list referenced"): + Workflow.from_dict( + { + "id": "w", + "steps": [ + {"id": "source", "action": "echo"}, + { + "id": "consumer", + "action": "echo", + "with": {"value": "{{ steps.source.output }}"}, + }, + ], + } + ) -class TestWorkflowCreation: - """Test workflow creation.""" - - @pytest.mark.workflow - def test_workflow_creation(self, mock_workflow): - """Test workflow creation.""" - assert mock_workflow.id == "workflow-1" - assert mock_workflow.name == "TestWorkflow" - - @pytest.mark.workflow - def test_workflow_status(self, mock_workflow): - """Test workflow status.""" - assert mock_workflow.status == "active" - - -class TestWorkflowExecution: - """Test workflow execution.""" - - @pytest.mark.workflow - def test_execute_workflow(self, mock_workflow): - """Test workflow execution.""" - result = mock_workflow.execute() - assert result["result"] == "success" - - @pytest.mark.workflow - def test_get_workflow_status(self, mock_workflow): - """Test getting workflow status.""" - status = mock_workflow.get_status() - assert status == "active" - - -class TestWorkflowConfiguration: - """Test workflow configuration.""" - - @pytest.mark.workflow - def test_workflow_config(self, mock_workflow_config): - """Test workflow configuration.""" - assert mock_workflow_config["name"] == "TestWorkflow" - assert len(mock_workflow_config["steps"]) == 2 - assert mock_workflow_config["enabled"] is True - - -class TestWorkflowEngine: - """Test workflow engine.""" - - @pytest.mark.engine - def test_create_workflow(self, mock_workflow_engine): - """Test creating workflow.""" - result = mock_workflow_engine.create_workflow() - assert result == "workflow-1" - - @pytest.mark.engine - def test_execute_workflow(self, mock_workflow_engine): - """Test executing workflow.""" - result = mock_workflow_engine.execute_workflow() - assert result["result"] == "success" - - @pytest.mark.engine - def test_list_workflows(self, mock_workflow_engine): - """Test listing workflows.""" - workflows = mock_workflow_engine.list_workflows() - assert isinstance(workflows, list) - - @pytest.mark.engine - def test_delete_workflow(self, mock_workflow_engine): - """Test deleting workflow.""" - result = mock_workflow_engine.delete_workflow() - assert result is True + with pytest.raises(WorkflowValidationError, match="invalid reference syntax"): + Workflow.from_dict( + { + "id": "w", + "steps": [{"id": "source", "action": "echo", "with": {"value": "{{ input }}"}}], + } + )