Skip to content

🧑‍🍳 Task Manager Engine (The Head Chef)

📌 Project Overview

The Task Manager Engine serves as the central processing runtime and distributed orchestration network of the application. Designed to act as the primary operational system of the infrastructure, it handles task queue consumption, orchestrates state machine lifecycles, dynamically targets processing lanes, and coordinates execution pipelines with modular business engines.

🏗️ Architecture Role & Station Topologies

The engine leverages a role-driven microstation configuration. By decoupling the execution logic from static paths, a single orchestration script can scale out horizontally into distinct runtime profiles using command-line arguments:

# Instance A: Launch as a Fast Pathway Controller (Consumes High-Speed Queue Traffic)
python task_manager/task_manager.py 1

# Instance B: Launch as a Slow/Dev Background Worker (Processes Heavy Batch Automation)
python task_manager/task_manager.py 2

Architectural Responsibilities

  • Decoupled Work Distribution: Offloads specific business operations to isolated helper extensions (modules/dev_engine1.py), protecting core orchestrator processes from script blockages.
  • Defensive Stream Ingestion: Collects data from socket connections in non-blocking, reliable 4KB block segments. This approach handles massive JSON payload variations without data truncation or socket dropping.
  • State Machine Lifecycle Coordination: Manages core state transitions across execution lifecycles, transforming tasks from Pending ➔ Running ➔ Completed.

🧬 Functional Processing Lifecycle Matrix

When an instance polls its assigned station pathway (1 or 2), it runs a continuous loop that executes the following operational steps:

[RS Layer Queue] │ ▼ (1. Fetch via call_rs) [Ingest & Validate Schema] ➔ (Updates Meta: Status="Running",Worker_ID, Start_Time)│ ├─► [Type 101: Admin/Dev] ──► Delegates to dev_engine.execute_admin_task()└─► [Other Tasks: Generic] ─► Routes to standard handling loops (5s isolation hold) │ ▼ (2. Update Payload & Compile Metrics)[State Assembly] ➔ (Appends Output Data, End_Time, Progress=100, Metrics, Error Logs) │ ▼ (3. Push Back to Core)[RS Completion Pipe] ➔ (Receives ACK_PUSHED or ACK_STORED confirmation)

### Detailed Lifecycle Properties Monitored

Operational Metric Derived Handling Context
task["meta"]["status"] Updated dynamically from Pending to Running upon ingest, and finally marked Completed.
task["meta"]["worker_id"] Injects explicit tracking strings (e.g., WORKER_1 or WORKER_2) to audit job histories.
task["meta"]["total_time"] Formulates processing time metrics by computing delta values between unix timestamp points.
task["meta"]["error_log"] Validates target variables (like fail_count) to track partial failures without crashing the runtime.

🛠️ Internal Network Command Architecture

The core network engine utilizes the call_rs(task_type, full_task) routing client to interface with the Receiving and Sending infrastructure via TCP sockets (localhost:7000):

  • Task Acquisition Requests: Fires raw string requests mapping to the active MY_STATION_TYPE code configuration to pop pending items from the queue.
  • Result Push Command (task_type: "3"): Invoked automatically when the worker payload finishes processing. This passes the full structured task schema back to the network hub to unblock synchronous clients or save the results safely to disk.

📊 Infrastructure Diagnostics & Logging

The orchestrator maintains detailed runtime tracing logs using append-only formatting handlers (task_manager.log): * Tracks task lifecycle operations, highlighting precise execution paths, source users, and processing performance intervals. * Captures structural connection anomalies, data stream parsing exceptions, and component pathing warnings.

🚀 Operational Deployment

1. Pre-Execution Requirements

Confirm your application directory structure includes the required backend modules package with appropriate initialization anchors:

Advetisement-offical/
├── modules/
│   ├── __init__.py
│   └── dev_engine1.py
└── task_manager/
    └── task_manager.py

2. Startup Invocation

Launch targeted orchestrator configurations directly from the repository root to establish connection streams:

python task_manager/task_manager.py 1