Resource Screener

Input Parameters

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

Execution Flow

Task Context

Before starting execution, initialize task_context.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(rundir, taskid): # Verify input files exist filestocheck = [ "layerbcandidates.json", "scan_plan.json" ]

missing = [] for f in filestocheck: 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(rundir, "layerbcandidates.json")) as f: candidatesdata = json.load(f) if "candidates" not in candidatesdata: return {"status": "failed", "error": "layerbcandidates.json missing 'candidates' array"} if len(candidatesdata["candidates"]) == 0: return {"status": "empty", "message": "No candidates after Layer B filtering"}

# Validate collector signal fields exist on first candidate first = candidatesdata["candidates"][0] requiredsignals = ["resourceid", "resourcetype", "entitytype"] missingsignals = [s for s in requiredsignals if s not in first] if missingsignals: return {"status": "failed", "error": f"candidates missing required fields: {missing_signals}"}

# Optional signal fields (collector may provide these) # cpuavgpct, networkmbperday, hashumanlogin, hasrealalert # ownerstatus, dependencies, reachability

with open(os.path.join(rundir, "scanplan.json")) as f: scanplan = json.load(f) if "dimensionframework" not in scanplan: return {"status": "failed", "error": "scanplan.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(rundir, "analysis"), existok=True)

return { "status": "success", "candidatescount": len(candidatesdata["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

DIMENSIONWEIGHTS = { "cpuidle": 0.35, "networkidle": 0.35, "ownershipclarity": 0.20, "data_sensitivity": 0.10 }

=== Signal → Dimension Score Converters ===

def computecpuidlescore(cpuavgpct): """Lower CPU peak = more idle = higher zombie suspicion. Returns (score, reliability).""" if cpuavgpct is None: return 0.0, 0.0 if cpuavgpct 0) force zombiescore=0.0 # b. cpuidle + networkidle both reliability=0.0 -> cap at 0.60 # c. hasrealalert=true suppresses networkidle dimension validdims = [d for d in scoreddims.values() if d["reliability"] > 0] if len(validdims) 0 else 0.0

# Gate b: cpuidle + networkidle both reliability=0 if cpurel == 0.0 and netrel == 0.0: zombiescore = min(zombiescore, 0.60)

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

zombiescore = min(max(zombiescore, 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 resourceid = candidate.get("resourceid", "").lower() if any(resourceid.startswith(p) for p in ["dr-", "backup-", "standby-"]): protectiontriggered = True if any((tags.get(k) or "") in ["standby", "backup"] for k in ["role", "usage"]): protection_triggered = True

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

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

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

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

ownerstatus = candidate.get("ownerstatus", "untagged") ownerbonuses = { "untagged": 0.10, "orphaned": 0.05, "activeowner": 0.0, "shared": 0.0 } ownerbonus = ownerbonuses.get(owner_status, 0.0)

priority = (zombiescore envmultiplier costweight) + ownerbonus priority = min(max(priority, 0.0), 1.0)

# Write intermediate result result = { "resourceid": resourceid, "resourcetype": resourcetype, "zombiescore": zombiescore, "suspectlevel": suspectlevel, "priority": priority, "protectiontriggered": protectiontriggered, "environment": environment, "scoreddimensions": scoreddims } results.append(result)

os.makedirs(os.path.join(rundir, "analysis"), existok=True) with open(os.path.join(rundir, "analysis", f"zombiesuspect{resourceid}.json"), "w") as f: json.dump(result, f, indent=2, ensure_ascii=False)

return { "status": "success", "candidatesprocessed": len(results), "highcount": sum(1 for r in results if r["suspectlevel"] == "high"), "mediumcount": sum(1 for r in results if r["suspectlevel"] == "medium"), "lowcount": 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
• resourcetype: compute/storage/network/database/cache/k8sworkload/k8sservice/k8sorphan/domain
• entity_type: Semantic entity type
• environment: prod/staging/dev/unknown
• creationtime: Mapped from candidate.provisionedat
• estimatedmonthlycost: Monthly cost (USD)
• suspect_level: high/medium/low
• zombie_score: 0-1 score (already calculated)
• priority: 0-1 ranking score (already calculated)
• ownerstatus: activeowner/orphaned/untagged/shared
• ownerdetail: "phase2pending" (populated by Phase 2)
• investigation_brief: 2-3 sentence summary
• blastradius: Blast radius analysis object containing summary (string), dependson (array), dependencies (array)
• evidence: Evidence list (each item with [CONFIRMED]/[SUPPORTED]/[INFERRED]/[UNKNOWN] marker)
• protectionruleschecked: Protection rules check results
• data_quality: Data quality record
• risk_factors: Risk factor list
• suggested_verification: Verification plan
• suggestednextstep: Recommended next action for this candidate

Forbidden fields:

• safetodelete
• verdict
• confidence (use zombie_score instead)
• recommendation
• blastradiusscore

Agent workflow:

1. Prepare the execution environment

1. Execute the agent with the prompt

1. Write results to:

• File: analysis/zombie_suspect_{resource_id}.json

Output

• Schema: schemas/suspect-resource-schema.json
• File: analysis/zombiesuspect{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 generateassessment(rundir, candidates): """Generate suspectassessment.json""" highsuspects = [c for c in candidates if c["suspectlevel"] == "high"] mediumsuspects = [c for c in candidates if c["suspectlevel"] == "medium"] lowsuspects = [c for c in candidates if c["suspect_level"] == "low"]

avgscore = sum(c["zombiescore"] for c in candidates) / len(candidates) if candidates else 0 totalsavings = sum(c["estimatedmonthlycost"] for c in highsuspects + medium_suspects)

# candidates[]: only 9 slim fields slimcandidates = [ { "resourceid": c["resourceid"], "resourcetype": c["resourcetype"], "suspectlevel": c["suspectlevel"], "zombiescore": c["zombiescore"], "priority": c["priority"], "estimatedmonthlycost": c["estimatedmonthlycost"], "ownerstatus": c["ownerstatus"], "investigationbrief": c["investigationbrief"], "suggestednextstep": c.get("suggestednext_step", "") } for c in candidates ]

assessment = { "assessedat": datetime.utcnow().isoformat() + "Z", "totalcandidates": len(candidates), "summary": { "highsuspect": len(highsuspects), "mediumsuspect": len(mediumsuspects), "lowsuspect": len(lowsuspects), "averagezombiescore": round(avgscore, 4), "totalestimatedmonthlysavings": totalsavings }, "selfcheck": { "passed": True, "gaps": [] }, "candidates": sorted( slim_candidates, key=lambda x: x["priority"], reverse=True ) }

return assessment

def run(rundir, taskid): # Read per-resource files generated by step 3 and build candidates list analysisdir = os.path.join(rundir, "analysis") candidates = [] for fname in sorted(os.listdir(analysisdir)): if fname.startswith("zombiesuspect") and fname.endswith(".json"): with open(os.path.join(analysisdir, 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 = generateassessment(rundir, candidates)

# Also load layerbcandidates.json to get estimatedmonthlycost for candidates that # may not have it in per-resource files layerbpath = os.path.join(rundir, "layerbcandidates.json") if os.path.exists(layerbpath): with open(layerbpath) as f: lbdata = json.load(f) lbmap = {c["resourceid"]: c for c in lbdata.get("candidates", [])} for c in assessment["candidates"]: if c.get("estimatedmonthlycost", 0) == 0: lbcandidate = lbmap.get(c["resourceid"], {}) c["estimatedmonthlycost"] = lbcandidate.get("estimatedmonthly_cost", 0)

# Recalculate totalsavings after cost enrichment assessment["summary"]["totalestimatedmonthlysavings"] = sum( c["estimatedmonthlycost"] for c in assessment["candidates"] if c["suspect_level"] in ("high", "medium") )

# Write suspectassessment.json assessmentpath = os.path.join(rundir, "analysis", "suspectassessment.json") with open(assessmentpath, "w") as f: json.dump(assessment, f, indent=2, ensureascii=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(rundir, taskid): assessmentpath = os.path.join(rundir, "analysis", "suspectassessment.json") if not os.path.exists(assessmentpath): return {"status": "failed", "error": "suspect_assessment.json not found — previous step must run first"}

with open(assessmentpath) as f: assessmentdata = json.load(f)

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

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

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

return {"status": "success", "outputfiles": [episodespath]}

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
• Source: 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.