# Resource Screener

> >-

- **Type:** Skill
- **Install:** `agentstack add skill-knoxops-open-devops-skills-resource-screener`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [KnoxOps](https://agentstack.voostack.com/s/knoxops)
- **Installs:** 0
- **Category:** [Cloud & Infrastructure](https://agentstack.voostack.com/c/cloud-infrastructure)
- **Latest version:** 0.1.0
- **License:** Apache-2.0
- **Upstream author:** [KnoxOps](https://github.com/KnoxOps)
- **Source:** https://github.com/KnoxOps/open-devops-skills/tree/master/plugins/ico/skills/resource-screener
- **Website:** https://knoxops.app?invite_token=GITHUB26

## Install

```sh
agentstack add skill-knoxops-open-devops-skills-resource-screener
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

## Input Parameters

| Name | Type | Required | Description |
|------|------|----------|-------------|
| run_dir | string | Yes | Work directory absolute path |
| ssh_key_path | string | No | SSH key path for SSH verification and cloud tags queries |
| task_id | string | Yes | Task ID for progress tracking |

## Execution Flow

### Task Context

Before starting execution, initialize `task_context.json`:

```json
{
  "task_id": "",
  "current_step": 0,
  "current_step_id": null,
  "status": "running",
  "steps": {
    "setup-layer-d-skeleton": "pending",
    "execute-scoring-pipeline": "pending",
    "generate-per-resource-files": "pending",
    "generate-suspect-assessment-json": "pending",
    "write-output-files": "pending"
  },
  "updated_at": ""
}
```

Update this file after each step completes. On error, set step status to `"failed"` and overall `status` to `"failed"`.

### Step 1: setup-layer-d-skeleton

**Type:** inline
**Description:** Input validation and initialization

## Execution
Follow these instructions:

import os
import json
from datetime import datetime

def run(run_dir, task_id):
  # Verify input files exist
  files_to_check = [
    "layer_b_candidates.json",
    "scan_plan.json"
  ]

  missing = []
  for f in files_to_check:
    path = os.path.join(run_dir, f)
    if not os.path.exists(path):
      missing.append(f)

  if missing:
    return {"status": "failed", "errors": f"Missing files: {missing}"}

  # Read and validate key fields
  try:
    with open(os.path.join(run_dir, "layer_b_candidates.json")) as f:
      candidates_data = json.load(f)
      if "candidates" not in candidates_data:
        return {"status": "failed", "error": "layer_b_candidates.json missing 'candidates' array"}
      if len(candidates_data["candidates"]) == 0:
        return {"status": "empty", "message": "No candidates after Layer B filtering"}

    # Validate collector signal fields exist on first candidate
    first = candidates_data["candidates"][0]
    required_signals = ["resource_id", "resource_type", "entity_type"]
    missing_signals = [s for s in required_signals if s not in first]
    if missing_signals:
      return {"status": "failed", "error": f"candidates missing required fields: {missing_signals}"}

    # Optional signal fields (collector may provide these)
    # cpu_avg_pct, network_mb_per_day, has_human_login, has_real_alert
    # owner_status, dependencies, reachability

    with open(os.path.join(run_dir, "scan_plan.json")) as f:
      scan_plan = json.load(f)
      if "dimension_framework" not in scan_plan:
        return {"status": "failed", "error": "scan_plan.json missing 'dimension_framework'"}
  except json.JSONDecodeError as e:
    return {"status": "failed", "error": f"JSON parse error: {e}"}

  # Create output directory
  os.makedirs(os.path.join(run_dir, "analysis"), exist_ok=True)

  return {
    "status": "success",
    "candidates_count": len(candidates_data["candidates"]),
    "prepared_at": datetime.utcnow().isoformat() + "Z"
  }

### Progress Tracking

After completing this step, update `task_context.json`:
- Set `current_step_id` to `"setup-layer-d-skeleton"`
- Set `steps.setup-layer-d-skeleton` to `"completed"`
### Step 2: execute-scoring-pipeline

**Type:** inline
**Description:** Execute full scoring pipeline on all candidates

## Execution
Follow these instructions:

import json
import os
import math

# Unified dimension weights (same for all resource types)
# Focus: CPU idleness, network idleness, ownership clarity, data sensitivity
# Heavier score on cpu+network near-zero = higher zombie suspicion
DIMENSION_WEIGHTS = {
  "cpu_idle": 0.35,
  "network_idle": 0.35,
  "ownership_clarity": 0.20,
  "data_sensitivity": 0.10
}

# === Signal → Dimension Score Converters ===

def compute_cpu_idle_score(cpu_avg_pct):
  """Lower CPU peak = more idle = higher zombie suspicion.
  Returns (score, reliability)."""
  if cpu_avg_pct is None:
    return 0.0, 0.0
  if cpu_avg_pct  0)  force zombie_score=0.0
    # b. cpu_idle + network_idle both reliability=0.0 -> cap at 0.60
    # c. has_real_alert=true suppresses network_idle dimension
    valid_dims = [d for d in scored_dims.values() if d["reliability"] > 0]
    if len(valid_dims)  0 else 0.0

      # Gate b: cpu_idle + network_idle both reliability=0
      if cpu_rel == 0.0 and net_rel == 0.0:
        zombie_score = min(zombie_score, 0.60)

      # Gate c: has_real_alert suppresses idle signal strength
      if candidate.get("has_real_alert", False):
        zombie_score = min(zombie_score, 0.70)

      zombie_score = min(max(zombie_score, 0.0), 1.0)

    # Step 3: Check protection rules (simplified - no SSH/crontab data)
    protection_triggered = False

    # DR/backup tag veto
    if isinstance(tags, str):
      try:
        tags = json.loads(tags)
      except:
        tags = {}
    purpose = (tags.get("purpose", "") or "").lower()
    if purpose in ["disaster-recovery", "backup"]:
      protection_triggered = True

    # Semantic naming pattern veto
    resource_id = candidate.get("resource_id", "").lower()
    if any(resource_id.startswith(p) for p in ["dr-", "backup-", "standby-"]):
      protection_triggered = True
    if any((tags.get(k) or "") in ["standby", "backup"] for k in ["role", "usage"]):
      protection_triggered = True

    # Step 4: Map suspect_level via environment-adjusted thresholds
    env_raw = candidate.get("environment", "").lower()
    if env_raw in ("prod", "production"):
      environment = "production"
    elif env_raw in ("staging", "stage"):
      environment = "staging"
    elif env_raw in ("dev", "development", "test", "testing"):
      environment = "dev"

    if protection_triggered:
      suspect_level = "low"
    else:
      env_thresholds = {
        "production": {"high": 0.90, "medium": 0.75},
        "staging": {"high": 0.80, "medium": 0.55},
        "dev": {"high": 0.75, "medium": 0.50}
      }
      thresholds = env_thresholds.get(environment, {"high": 0.90, "medium": 0.75})
      if zombie_score >= thresholds["high"]:
        suspect_level = "high"
      elif zombie_score >= thresholds["medium"]:
        suspect_level = "medium"
      else:
        suspect_level = "low"

    # Step 5: Compute priority
    env_multipliers = {
      "production": 0.6,
      "staging": 0.8, "stage": 0.8,
      "dev": 1.0, "development": 1.0, "test": 1.0, "testing": 1.0
    }
    env_multiplier = env_multipliers.get(environment, 0.6)

    estimated_cost = candidate.get("estimated_monthly_cost", 0)
    if estimated_cost <= 0:
      cost_weight = 0.30
    else:
      cost_weight = math.log10(estimated_cost + 1) / 3
      cost_weight = min(max(cost_weight, 0.30), 1.0)

    owner_status = candidate.get("owner_status", "untagged")
    owner_bonuses = {
      "untagged": 0.10,
      "orphaned": 0.05,
      "active_owner": 0.0,
      "shared": 0.0
    }
    owner_bonus = owner_bonuses.get(owner_status, 0.0)

    priority = (zombie_score * env_multiplier * cost_weight) + owner_bonus
    priority = min(max(priority, 0.0), 1.0)

    # Write intermediate result
    result = {
      "resource_id": resource_id,
      "resource_type": resource_type,
      "zombie_score": zombie_score,
      "suspect_level": suspect_level,
      "priority": priority,
      "protection_triggered": protection_triggered,
      "environment": environment,
      "scored_dimensions": scored_dims
    }
    results.append(result)

    os.makedirs(os.path.join(run_dir, "analysis"), exist_ok=True)
    with open(os.path.join(run_dir, "analysis", f"zombie_suspect_{resource_id}.json"), "w") as f:
      json.dump(result, f, indent=2, ensure_ascii=False)

  return {
    "status": "success",
    "candidates_processed": len(results),
    "high_count": sum(1 for r in results if r["suspect_level"] == "high"),
    "medium_count": sum(1 for r in results if r["suspect_level"] == "medium"),
    "low_count": sum(1 for r in results if r["suspect_level"] == "low")
  }

### Progress Tracking

After completing this step, update `task_context.json`:
- Set `current_step_id` to `"execute-scoring-pipeline"`
- Set `steps.execute-scoring-pipeline` to `"completed"`
### Step 3: generate-per-resource-files

**Type:** agent
**Description:** Generate per-resource output files (LLM assistant to produce investigation_brief etc.)

## Execution
Launch an independent agent with the following prompt file:

**Dispatch instruction:**

For each suspect candidate, generate a `analysis/zombie_suspect_{resource_id}.json` file based on the following information.

Required fields:
- assessed_at: ISO 8601 timestamp
- resource_id: Unique resource identifier
- resource_type: compute/storage/network/database/cache/k8s_workload/k8s_service/k8s_orphan/domain
- entity_type: Semantic entity type
- environment: prod/staging/dev/unknown
- creation_time: Mapped from candidate.provisioned_at
- estimated_monthly_cost: Monthly cost (USD)
- suspect_level: high/medium/low
- zombie_score: 0-1 score (already calculated)
- priority: 0-1 ranking score (already calculated)
- owner_status: active_owner/orphaned/untagged/shared
- owner_detail: "phase2_pending" (populated by Phase 2)
- investigation_brief: 2-3 sentence summary
- blast_radius: Blast radius analysis object containing summary (string), depends_on (array), dependencies (array)
- evidence: Evidence list (each item with [CONFIRMED]/[SUPPORTED]/[INFERRED]/[UNKNOWN] marker)
- protection_rules_checked: Protection rules check results
- data_quality: Data quality record
- risk_factors: Risk factor list
- suggested_verification: Verification plan
- suggested_next_step: Recommended next action for this candidate

Forbidden fields:
- safe_to_delete
- verdict
- confidence (use zombie_score instead)
- recommendation
- blast_radius_score

**Agent workflow:**

1. Prepare the execution environment

2. Execute the agent with the prompt

3. Write results to:
   - File: `analysis/zombie_suspect_{resource_id}.json`

## Output
- **Schema:** schemas/suspect-resource-schema.json
  - **File:** analysis/zombie_suspect_{resource_id}.json

### Progress Tracking

After completing this step, update `task_context.json`:
- Set `current_step_id` to `"generate-per-resource-files"`
- Set `steps.generate-per-resource-files` to `"completed"`
### Step 4: generate-suspect-assessment-json

**Type:** inline
**Description:** Generate summary output file

## Input Files
- `analysis/zombie_suspect_{resource_id}.json` (from Step generate-per-resource-files, schema: schemas/suspect-resource-schema.json)

## Execution
Follow these instructions:

import json
import os
from datetime import datetime

def generate_assessment(run_dir, candidates):
  """Generate suspect_assessment.json"""
  high_suspects = [c for c in candidates if c["suspect_level"] == "high"]
  medium_suspects = [c for c in candidates if c["suspect_level"] == "medium"]
  low_suspects = [c for c in candidates if c["suspect_level"] == "low"]

  avg_score = sum(c["zombie_score"] for c in candidates) / len(candidates) if candidates else 0
  total_savings = sum(c["estimated_monthly_cost"] for c in high_suspects + medium_suspects)

  # candidates[]: only 9 slim fields
  slim_candidates = [
    {
      "resource_id": c["resource_id"],
      "resource_type": c["resource_type"],
      "suspect_level": c["suspect_level"],
      "zombie_score": c["zombie_score"],
      "priority": c["priority"],
      "estimated_monthly_cost": c["estimated_monthly_cost"],
      "owner_status": c["owner_status"],
      "investigation_brief": c["investigation_brief"],
      "suggested_next_step": c.get("suggested_next_step", "")
    }
    for c in candidates
  ]

  assessment = {
    "assessed_at": datetime.utcnow().isoformat() + "Z",
    "total_candidates": len(candidates),
    "summary": {
      "high_suspect": len(high_suspects),
      "medium_suspect": len(medium_suspects),
      "low_suspect": len(low_suspects),
      "average_zombie_score": round(avg_score, 4),
      "total_estimated_monthly_savings": total_savings
    },
    "self_check": {
      "passed": True,
      "gaps": []
    },
    "candidates": sorted(
      slim_candidates,
      key=lambda x: x["priority"],
      reverse=True
    )
  }

  return assessment

def run(run_dir, task_id):
  # Read per-resource files generated by step 3 and build candidates list
  analysis_dir = os.path.join(run_dir, "analysis")
  candidates = []
  for fname in sorted(os.listdir(analysis_dir)):
    if fname.startswith("zombie_suspect_") and fname.endswith(".json"):
      with open(os.path.join(analysis_dir, fname)) as f:
        candidates.append(json.load(f))

  if not candidates:
    return {"status": "failed", "error": "No per-resource analysis files found from step 3"}

  assessment = generate_assessment(run_dir, candidates)

  # Also load layer_b_candidates.json to get estimated_monthly_cost for candidates that
  # may not have it in per-resource files
  layer_b_path = os.path.join(run_dir, "layer_b_candidates.json")
  if os.path.exists(layer_b_path):
    with open(layer_b_path) as f:
      lb_data = json.load(f)
    lb_map = {c["resource_id"]: c for c in lb_data.get("candidates", [])}
    for c in assessment["candidates"]:
      if c.get("estimated_monthly_cost", 0) == 0:
        lb_candidate = lb_map.get(c["resource_id"], {})
        c["estimated_monthly_cost"] = lb_candidate.get("estimated_monthly_cost", 0)

  # Recalculate total_savings after cost enrichment
  assessment["summary"]["total_estimated_monthly_savings"] = sum(
    c["estimated_monthly_cost"]
    for c in assessment["candidates"]
    if c["suspect_level"] in ("high", "medium")
  )

  # Write suspect_assessment.json
  assessment_path = os.path.join(run_dir, "analysis", "suspect_assessment.json")
  with open(assessment_path, "w") as f:
    json.dump(assessment, f, indent=2, ensure_ascii=False)

  return {
    "status": "success",
    "data": assessment,
    "candidates_assessed": len(candidates)
  }

Write the output to the specified output file.

## Output
- **Schema:** schemas/assessment-schema.json
  - **File:** analysis/suspect_assessment.json

### Progress Tracking

After completing this step, update `task_context.json`:
- Set `current_step_id` to `"generate-suspect-assessment-json"`
- Set `steps.generate-suspect-assessment-json` to `"completed"`
### Step 5: write-output-files

**Type:** inline
**Description:** Write final progress tracking output

## Execution
Follow these instructions:

import json
import os
from datetime import datetime

def run(run_dir, task_id):
  assessment_path = os.path.join(run_dir, "analysis", "suspect_assessment.json")
  if not os.path.exists(assessment_path):
    return {"status": "failed", "error": "suspect_assessment.json not found — previous step must run first"}

  with open(assessment_path) as f:
    assessment_data = json.load(f)

  # Update progress file
  episodes_path = os.path.join(run_dir, "scan_episodes.json")
  try:
    with open(episodes_path) as f:
      episodes = json.load(f)
  except:
    episodes = {"events": []}

  episodes["events"].append({
    "timestamp": datetime.utcnow().isoformat() + "Z",
    "layer": "Layer D",
    "status": "completed",
    "candidates_assessed": assessment_data["summary"]["total_candidates"]
  })

  with open(episodes_path, "w") as f:
    json.dump(episodes, f, indent=2, ensure_ascii=False)

  return {"status": "success", "output_files": [episodes_path]}

### Progress Tracking

After completing this step, update `task_context.json`:
- Set `current_step_id` to `"write-output-files"`
- Set `steps.write-output-files` to `"completed"`

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [KnoxOps](https://github.com/KnoxOps)
- **Source:** [KnoxOps/open-devops-skills](https://github.com/KnoxOps/open-devops-skills)
- **License:** Apache-2.0
- **Homepage:** https://knoxops.app?invite_token=GITHUB26

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** yes
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-knoxops-open-devops-skills-resource-screener
- Seller: https://agentstack.voostack.com/s/knoxops
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.