Skip to content

Adapt Rq

Adapt RQ — mvp.adapt_rq

Cluster: Data Processing | Type: component | MCP Tools: 36

Overview

Redis Queue (RQ) adapter that enqueues Python functions as background jobs, checks job status, cancels jobs, lists active workers, and snapshots queue depth without blocking the calling agent. The Tier-1 block carries the canonical reliability envelope (completion_state, warning_card, evidence, request_id, run_id) additively on its dict-based output; the MCP block adds SQLite persistence for job metadata, result caching, queue statistics, multi-queue management, and a constrained backend-native adapter family.

The Tier-1 block (AdaptRQBlock) is fail-closed: completion_state "verified" is reachable only when a real RQ/Redis operation executed and returned concrete backend data (e.g. a job_id, status, worker list, or queue length). When the optional rq/redis libraries are missing or unimportable (including the Windows fork limitation) the backend-unavailable path returns Result.fail with blocked-escalated (G6_E_RQ_BACKEND_UNAVAILABLE) — it does NOT read as a successful degraded-but-OK result. A connection/operation exception fails closed to blocked-escalated (G6_E_RQ_CONNECTION_FAILED); bad input, unknown op, invalid args JSON, and job-not-found fail closed to qualified-draft (typed G6_E_RQ_INVALID_INPUT / G6_E_RQ_UNKNOWN_OP / G6_E_RQ_INVALID_ARGS_JSON / G6_E_RQ_JOB_NOT_FOUND). No fallback queue is materialised. The block stays experimental (no benchmark validation).

When Redis is not reachable or the optional rq/redis libraries are missing, Redis-backed MCP ops return a canonical reliability envelope with completion_state, warning_card, evidence, request_id, run_id, and side_effect_committed; no fallback queue is materialised. Missing required native backends and unsafe native mutations use blocked-escalated, while SQLite/default fallback paths use qualified-draft. Call rq_readiness or rq_native_capability_info first to inspect dependency status, Redis reachability, worker visibility, queue visibility, SQLite fallback availability, native allowlists, gate requirements, and disabled reasons.

When to use:

  • Offloading long-running computation (ML training, media transcoding) to a background worker pool
  • Checking the status of previously submitted jobs without polling Redis directly
  • Managing and monitoring multiple named queues from an orchestration agent
  • Inspecting allowlisted native RQ queue/job/registry/worker metadata with explicit degradation state
  • Integrating asynchronous task dispatch into a pipeline that must remain responsive

Operational warning:

adapt_rq persists job metadata and native audit summaries in SQLite, but durable execution still depends on Redis and an RQ worker process. Components such as recursive_architect can enqueue work through adapt_rq; those jobs only keep running across MCP restarts if Redis is reachable and a worker is listening on the target queue. Without that worker, callers may record metadata or receive degraded snapshots, but no background work will complete.

For Recursive Architect durable runs, start:

python -m rq worker g6-architect --url redis://localhost:6379/0

Example:

from mvp.adapt_rq import AdaptRQBlock, RQInput

block = AdaptRQBlock(name="rq")
result = block.infer(RQInput(op="ops"))
# result.value -> RQOutput listing all Tier-1 ops the block supports.

# For real enqueue (REQUIRES Redis running locally on default port 6379):
#   block.infer(RQInput(
#       op="enqueue",
#       queue_name="ml_jobs",
#       func_path="myapp.tasks.train_model",
#       args_json='["dataset_v3"]',
#       kwargs_json='{"epochs": 50}',
#   ))  # -> result.value["job_id"] is the job ID for status polling

Works well with: adapt_sklearn, adapt_pytorch, adapt_keras

Public API

RQOutput

Tier 1 output - dict wrapper with canonical reliability envelope.

Constructor:

Parameter Type Default
data dict[str, Any] required
degraded bool False
degradation_reason str ''
completion_state ReliabilityLabel 'qualified-draft'
warning_card dict[str, Any] \| None None
evidence list[dict[str, Any]] \| None None
request_id str ''
task_id str ''
run_id str ''

Methods:

get(key: str, default: Any = None) -> Any

model_dump() -> dict[str, Any]

AdaptRQBlock(AIBlock['RQInput', 'RQOutput', None])

Tier 1 RQ block - basic queue ops with resource guardrails.

Field Type Default
name str 'adapt_rq'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)
redis_url str ''

Methods:

infer(data: RQInput) -> Result[RQOutput]

RQInput(BaseModel)

Tier 1 input for basic RQ operations.

Field Type Default
op RQOp required
queue_name str 'default'
func_path str ''
args_json str '[]'
kwargs_json str '{}'
job_id str ''
scheduled_at str ''
delay_seconds int 0
request_id str ''
task_id str ''
run_id str ''

AdaptRQMCPBlock(AIBlock['MCPRQInput', 'MCPRQOutput', dict])

35-op RQ MCP block with SQLite persistence and Redis backend.

Field Type Default
name str 'adapt_rq_mcp'
state dict field(default_factory=dict)
db_path str field(default_factory=lambda: os.environ.get('RQ_DB_PATH', _DEFAULT_DB))
redis_url str field(default_factory=lambda: os.environ.get('REDIS_URL', 'redis://localhost:6379/0'))
resource_bounds ResourceBounds field(default_factory=ResourceBounds)
usage ResourceUsage field(default_factory=ResourceUsage)
agentic_planner RecoveryPlanner \| None None

Methods:

infer(data: MCPRQInput) -> Result[MCPRQOutput]

MCPRQInput(BaseModel)

Input for all RQ MCP operations.

Field Type Default
op MCPRQOp required
queue_name str 'default'
func_path str ''
args_json str '[]'
kwargs_json str '{}'
jobs_json str '[]'
scheduled_at str ''
delay_seconds int 0
job_id str ''
worker_name str ''
status str ''
result_json str ''
error_message str ''
duration_seconds float 0.0
tags str ''
min_frequency int 1
limit int 50
query str ''
top_k int 10
dlq_queue_name str 'dlq'
attempt_count int 0
max_attempts int 3
native_object str ''
native_action str ''
native_payload str '{}'
permission_token str ''
request_id str ''
task_id str ''
run_id str ''

MCPRQOutput(BaseModel)

Output for all RQ MCP operations.

Field Type Default
op str ''
job_id str ''
job_ids list[str] Field(default_factory=list)
status str ''
func_name str ''
enqueued_at str ''
started_at str ''
ended_at str ''
scheduled_at str ''
result_json str ''
cancelled bool False
deleted bool False
requeued bool False
recorded bool False
found bool False
count int 0
removed_count int 0
workers list[dict[str, Any]] Field(default_factory=list)
records list[dict[str, Any]] Field(default_factory=list)
patterns list[dict[str, Any]] Field(default_factory=list)
scores list[float] Field(default_factory=list)
pending int 0
started int 0
finished int 0
failed int 0
message str ''
metadata dict[str, Any] Field(default_factory=dict)
degraded bool False
degradation_reason str ''
request_id str ''
task_id str ''
run_id str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict[str, Any] Field(default_factory=dict)
evidence list[dict[str, Any]] Field(default_factory=list)
side_effect_committed bool \| Literal['unknown'] False
redis_reachable bool False
declared_queues list[str] Field(default_factory=list)
dependency_status dict[str, bool] Field(default_factory=dict)

MCP Tools

Operation Source
enqueue rq_mcp
enqueue_many rq_mcp
enqueue_at rq_mcp
enqueue_in rq_mcp
empty_queue rq_mcp
job_status rq_mcp
job_result rq_mcp
cancel_job rq_mcp
delete_job rq_mcp
retry_job rq_mcp
list_started rq_mcp
list_finished rq_mcp
list_failed rq_mcp
list_deferred rq_mcp
list_scheduled rq_mcp
requeue_failed rq_mcp
list_workers rq_mcp
worker_status rq_mcp
worker_count rq_mcp
record_job rq_mcp
query_jobs rq_mcp
get_job_patterns rq_mcp
queue_snapshot rq_mcp
search rq_mcp
info rq_mcp
move_to_dlq rq_mcp
rq_readiness rq_mcp
recommend_recovery rq_mcp
list_patterns rq_mcp
native_capability_info rq_mcp
native_queue_inspect rq_mcp
native_job_inspect rq_mcp
native_registry_inspect rq_mcp
native_worker_inspect rq_mcp
native_mutation_request rq_mcp
unknown rq_mcp