diegoecab/oci-deal-accelerator
1
1
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Auto-refresh KB pricing, align skill with Anthropic best practices

Browse Source 
Why this set of changes:
- KB pricing was drifting silently — domain files (database.yaml,
  storage.yaml, etc.) had prices 30-800% off the live Oracle API and
  nobody read them. The skill was auditing as stale on every check
  with no path to fix it.
- The skill itself violated Anthropic's spec (`name` field had
  uppercase/spaces) and was over the 500-line guideline (647 lines),
  hurting discovery and load performance.
- Welcome flow occasionally improvised the menu instead of reading
  SKILL.md, missing options.

Pricing — single source of truth, fully automated:
- Extend tools/refresh_sku_catalog.py with --refresh-domain compute,
  pulls shape-level prices from the Oracle public pricing API
  (apexapps.oracle.com), preserves manual fields (notes, GPU specs,
  free-tier annotations, estimation_helpers), recomputes derived
  monthly values, and protects $0 free-tier prices from overwrite.
- Delete 12 redundant pricing/<domain>.yaml files. They duplicated
  oci-sku-catalog.yaml with worse abstractions and were nobody's
  source of truth (no tool consumed them).
- Migrate the genuinely valuable knowledge from those 12 files
  (billing models, BYOL rules, free-tier rules, ECPU vs OCPU,
  X11M elastic model, hyperscaler comparisons, service nuances)
  into kb/field-knowledge/pricing-knowledge.yaml — non-numeric,
  no refresh needed.
- Result: pricing freshness check goes from 13 stale files to 0.

KB freshness automation:
- Add tools/kb_freshness.py — wrapper around kb_linter.check_freshness()
  with --check, --auto-refresh, --json, --quiet modes. Bridges stale
  files to their refresh tools (SKU catalog, compute domain, arch
  center). Wired into the welcome flow as a pre-flight banner that
  asks the user before refreshing.
- Fix pre-existing kb_linter bug: it crashed on the 45 multi-doc
  YAML files (frontmatter + body pattern) because it used safe_load
  instead of safe_load_all. Freshness check was effectively dead.
- Standardize timestamp field: linter now accepts last_verified,
  last_updated, and last_refreshed; refresh_arch_catalog writes
  last_verified instead of last_refreshed.
- Add make freshness / make freshness-refresh targets.

Skill alignment with Anthropic Agent Skills best practices:
- Rename `name: OCI Deal Accelerator` to `oci-deal-accelerator`
  to comply with the [a-z0-9-]{1,64} spec.
- Refactor SKILL.md from 674 to 445 lines via progressive disclosure:
  extract WA review output format, ECAL readiness format, and output
  conventions into docs/skill/*.md referenced from the main file.
- Add scripts/sync-skill.py + make sync-skill: source of truth is
  root SKILL.md, .agents/skills/oci-deal-accelerator/SKILL.md is
  auto-generated. make lint validates sync.
- Add evaluations/ with 3 manual baseline scenarios (welcome-flow,
  full-proposal, wa-review) per the Anthropic best-practices guidance
  to "build evaluations first."

Welcome flow hardening:
- Tighten CLAUDE.md to MANDATE reading SKILL.md before showing the
  menu (no improvising), and document the freshness pre-flight check
  with the ask-before-refresh user flow.
- Update SKILL.md welcome flow to instruct: parse kb_freshness JSON,
  show banner with stale count + oldest file, prompt user to refresh
  (only when an automated tool exists), fall back silently on errors.

Linter hygiene (zero remaining issues):
- Expand config/kb-tags.yaml taxonomy with features, operations,
  metrics, limitations sections covering 31 previously-unknown tags
  used in field findings (rac, ecpu, refreshable-clone, hnsw, etc.).
- Assign owners for kb/compatibility/, kb/competitive/,
  kb/well-architected/ (Diego Cabrera as default until team grows);
  kb/pricing/ marked as "Auto-refreshed" since it no longer needs
  human ownership.
- kb_linter accepts top-level `date` as fallback for contributor
  block; migrate FF-202603-008 from legacy `reported_by` to
  contributor block.
- Result: linter goes from 45 issues to 0.

Other:
- Recompute estimation_helpers monthly values in compute.yaml after
  the price refresh (they were derived from the old E5/A1 numbers).
- Add kb/README.md — contributor guide (directory map, frontmatter
  spec, refresh tooling, review cadence).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-04-08 23:59:32 -03:00
parent 69b0ccb4b8
commit ca93a0aa4e
37 changed files with 1689 additions and 1639 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
tools/kb_freshness.py Normal file
View File

@@ -0,0 +1,176 @@
#!/usr/bin/env python3
"""
OCI Deal Accelerator — KB Freshness Reporter
Thin wrapper around kb_linter.check_freshness() that:
- reports stale/warning KB files in a compact, scriptable format
- knows which stale files have a refresh tool, and can run it on demand
- is safe to call from the skill welcome flow as a precondition check
Usage:
python tools/kb_freshness.py --check # human-readable report
python tools/kb_freshness.py --check --quiet # no output, exit 1 if stale
python tools/kb_freshness.py --check --json # JSON output for the skill
python tools/kb_freshness.py --auto-refresh # run refresh tools for stale files
Exit codes:
0 — all files fresh (or all stale files refreshed successfully)
1 — stale files found (in --check mode) or refresh failed (--auto-refresh)
2 — internal error
"""
import argparse
import json
import os
import subprocess
import sys
from typing import Any, Dict, List, Optional
# Reuse the linter's freshness implementation. kb_linter.py lives in the same
# directory; add it to sys.path so this script can be invoked from anywhere.
_HERE = os.path.dirname(os.path.realpath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from kb_linter import check_freshness # type: ignore # noqa: E402
PROJECT_ROOT = os.path.dirname(_HERE)
# Bridge: KB file -> refresh command (relative to PROJECT_ROOT).
# Only files with a known automated source belong here. Files with no entry
# can only be refreshed by a human and will be reported as such.
REFRESH_TOOLS: Dict[str, List[str]] = {
"kb/pricing/oci-sku-catalog.yaml": [
sys.executable, "tools/refresh_sku_catalog.py", "--refresh",
],
"kb/pricing/compute.yaml": [
sys.executable, "tools/refresh_sku_catalog.py", "--refresh-domain", "compute",
],
"kb/architecture-center/catalog.yaml": [
sys.executable, "tools/refresh_arch_catalog.py", "--whats-new",
],
}
def collect_stale() -> List[Dict[str, Any]]:
"""Return only files in WARNING or STALE state, sorted oldest first."""
results = check_freshness()
stale = [r for r in results if r["status"] in ("STALE", "WARNING")]
stale.sort(key=lambda r: r["age_days"], reverse=True)
return stale
def has_refresh_tool(rel_path: str) -> bool:
return rel_path in REFRESH_TOOLS
def run_refresh(rel_path: str) -> bool:
"""Run the refresh tool for `rel_path`. Returns True on success."""
cmd = REFRESH_TOOLS[rel_path]
print(f" refreshing {rel_path} ...")
proc = subprocess.run(cmd, cwd=PROJECT_ROOT)
return proc.returncode == 0
def cmd_check(args: argparse.Namespace) -> int:
stale = collect_stale()
if args.json:
payload = {
"stale_count": len(stale),
"files": [
{
"file": s["file"],
"status": s["status"],
"age_days": s["age_days"],
"refreshable": has_refresh_tool(s["file"]),
}
for s in stale
],
}
print(json.dumps(payload, indent=2))
return 1 if stale else 0
if not stale:
if not args.quiet:
print("KB freshness: OK (all files within thresholds)")
return 0
if args.quiet:
return 1
refreshable = sum(1 for s in stale if has_refresh_tool(s["file"]))
print(f"KB freshness: {len(stale)} file(s) need attention "
f"({refreshable} have an automatic refresh tool)")
print()
for s in stale:
marker = "[auto]" if has_refresh_tool(s["file"]) else "[manual]"
print(f" {s['status']:<7} {marker:<8} {s['age_days']:>4}d {s['file']}")
print()
print("To refresh files with [auto]: python tools/kb_freshness.py --auto-refresh")
return 1
def cmd_auto_refresh(args: argparse.Namespace) -> int:
stale = collect_stale()
refreshable = [s for s in stale if has_refresh_tool(s["file"])]
if not refreshable:
print("KB freshness: nothing to auto-refresh "
f"({len(stale)} stale file(s), none have a refresh tool)")
return 0 if not stale else 1
print(f"Auto-refreshing {len(refreshable)} file(s)...")
failures: List[str] = []
for s in refreshable:
if not run_refresh(s["file"]):
failures.append(s["file"])
if failures:
print(f"\nFAILED: {len(failures)} refresh(es) errored:")
for f in failures:
print(f" - {f}")
return 1
print("\nAll automatic refreshes completed.")
# Re-check to confirm
remaining = [s for s in collect_stale() if has_refresh_tool(s["file"])]
if remaining:
print(f"WARNING: {len(remaining)} file(s) still stale after refresh:")
for r in remaining:
print(f" - {r['file']} ({r['age_days']}d)")
return 1
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="kb_freshness",
description="Report and refresh stale KB files.",
)
parser.add_argument("--check", action="store_true",
help="Report stale files (default mode).")
parser.add_argument("--auto-refresh", action="store_true",
help="Run refresh tools for stale files that support them.")
parser.add_argument("--quiet", action="store_true",
help="Suppress output; use exit code only.")
parser.add_argument("--json", action="store_true",
help="Emit JSON (for the skill welcome flow to consume).")
return parser
def main() -> int:
args = build_parser().parse_args()
if args.auto_refresh:
return cmd_auto_refresh(args)
# Default mode is --check
return cmd_check(args)
if __name__ == "__main__":
try:
sys.exit(main())
except Exception as e: # pragma: no cover
print(f"kb_freshness: internal error: {e}", file=sys.stderr)
sys.exit(2)

23
tools/kb_linter.py
View File

@@ -60,7 +60,14 @@ def _load_yaml(path: str) -> Optional[Dict[str, Any]]:
if not os.path.isfile(resolved):
return None
with open(resolved, "r", encoding="utf-8") as fh:
return yaml.safe_load(fh)
# Many KB files use a frontmatter pattern: a metadata doc followed by
# a body doc, separated by `---`. Merge all dict docs into one so
# callers see a single namespace (later docs override earlier keys).
merged: Dict[str, Any] = {}
for doc in yaml.safe_load_all(fh):
if isinstance(doc, dict):
merged.update(doc)
return merged or None
def _load_governance() -> Dict[str, Any]:
@@ -125,8 +132,14 @@ def check_contributor_blocks(tracker_path: str = TRACKER_PATH) -> List[str]:
else:
issues.append(f"{fid}: missing contributor block")
continue
# The finding's top-level `date` is the canonical discovery date and
# acts as a fallback when the contributor block doesn't repeat it.
finding_date = f.get("date")
for field in required:
if not contributor.get(field):
value = contributor.get(field)
if not value and field == "date" and finding_date:
continue
if not value:
issues.append(f"{fid}: contributor missing required field '{field}'")
conf = contributor.get("confidence", "")
if conf and conf not in valid_confidence:
@@ -239,7 +252,11 @@ def check_freshness() -> List[Dict[str, Any]]:
if not data:
continue
last_verified = data.get("last_verified") or data.get("last_updated")
last_verified = (
data.get("last_verified")
or data.get("last_updated")
or data.get("last_refreshed")
)
if last_verified:
age = _days_ago(last_verified)
if age is not None:

2
tools/refresh_arch_catalog.py
View File

@@ -314,7 +314,7 @@ def main():
existing_urls.add(url)
new_count += 1
catalog["last_refreshed"] = datetime.now().strftime("%Y-%m-%d")
catalog["last_verified"] = datetime.now().strftime("%Y-%m-%d")
catalog["source"] = "https://docs.oracle.com/en/solutions/oracle-architecture-center/"
catalog["entry_count"] = len(catalog.get("entries", []))
save_catalog(catalog)

264
tools/refresh_sku_catalog.py
View File

@@ -28,6 +28,7 @@ Usage:
import argparse
import json
import os
import re
import sys
import time
from datetime import date
@@ -50,6 +51,17 @@ DEFAULT_CURRENCY = "USD"
# ── Paths ─────────────────────────────────────────────────────────
TOOLS_DIR = Path(__file__).resolve().parent
CATALOG_PATH = TOOLS_DIR.parent / "kb" / "pricing" / "oci-sku-catalog.yaml"
COMPUTE_DOMAIN_PATH = TOOLS_DIR.parent / "kb" / "pricing" / "compute.yaml"
# ── Domain refresh registry ───────────────────────────────────────
# Maps a domain key (used by --refresh-domain) to its file path and
# refresher function. Add a new entry here to support more domains.
DOMAIN_REGISTRY = {
"compute": {
"path": COMPUTE_DOMAIN_PATH,
"refresher": "refresh_compute_yaml",
},
}
def fetch_all_products(currency="USD", verbose=False):
@@ -293,6 +305,252 @@ def inspect_sku(part_number):
))
# ── Domain refresh: compute ───────────────────────────────────────
# Compute family token regex. Matches the family token (E3-E6, A1/A2/A4, X9)
# inside shape names like VM.Standard.E6.Flex, BM.Standard.E5, BM.Standard.x9.
_COMPUTE_FAMILY_RE = re.compile(r'\b(E[3-6]|A[1-4]|X9|x9)\b')
def _shape_family(shape_name):
"""Return the family token (E5, A2, X9, ...) used in API displayName,
or None if the shape can't be mapped automatically.
Special case: VM.Optimized3.Flex maps to the 'Optimized X9' product line.
"""
if "Optimized3" in shape_name or "Optimized" in shape_name:
return "X9_OPTIMIZED"
m = _COMPUTE_FAMILY_RE.search(shape_name)
if m:
return m.group(1).upper()
return None
def build_compute_family_lookup(api_map):
"""Scan the API for Compute Standard / Optimized OCPU and Memory products,
return {family: {'ocpu': price, 'memory': price, 'ocpu_sku': sku, 'memory_sku': sku}}.
"""
lookup = {}
for sku, product in api_map.items():
name = product.get("displayName", "")
if "Compute" not in name:
continue
if "Cloud@Customer" in name:
continue
if "Dense I/O" in name or "Dense IO" in name:
continue # different pricing tier
if "HPC" in name:
continue
if "VMware" in name:
continue
if "GPU" in name:
continue # GPU shapes not handled in this domain refresher
is_optimized = "Optimized" in name
is_standard = "Standard" in name
if not (is_optimized or is_standard):
continue
# Find the family token in the displayName
m = _COMPUTE_FAMILY_RE.search(name)
if not m:
continue
family = m.group(1).upper()
if is_optimized:
family = "X9_OPTIMIZED"
metric = product.get("metricName", "")
price = extract_payg_price(product)
slot = lookup.setdefault(family, {})
if "OCPU" in metric.upper():
# Prefer the first match (deterministic across runs since API order is stable)
slot.setdefault("ocpu", price)
slot.setdefault("ocpu_sku", sku)
elif "GIGABYTE" in metric.upper() or "Gigabytes" in metric:
slot.setdefault("memory", price)
slot.setdefault("memory_sku", sku)
return lookup
def _load_multidoc_yaml(path):
"""Load a multi-doc YAML file (frontmatter + body) and return both as dicts."""
with open(str(path), "r", encoding="utf-8") as fh:
docs = list(yaml.safe_load_all(fh))
frontmatter = {}
body = {}
for d in docs:
if not isinstance(d, dict):
continue
# Frontmatter typically has last_verified, source, description, currency
if "last_verified" in d or "source" in d or "currency" in d:
frontmatter.update(d)
else:
body.update(d)
return frontmatter, body
def _save_multidoc_yaml(path, frontmatter, body):
"""Write frontmatter + body as a two-doc YAML file."""
with open(str(path), "w", encoding="utf-8") as fh:
yaml.safe_dump(frontmatter, fh, default_flow_style=False, sort_keys=False, allow_unicode=True)
fh.write("---\n")
yaml.safe_dump(body, fh, default_flow_style=False, sort_keys=False, allow_unicode=True, width=120)
def refresh_compute_yaml(verbose=False):
"""Refresh kb/pricing/compute.yaml from the live API.
Updates ocpu_per_hour, memory_per_gb_hour, and monthly_730h for shapes
in flexible_shapes and bare_metal_shapes. Skips GPU shapes (out of scope).
Preserves notes, free_tier, gpu_model, and other manual fields.
"""
print("Fetching current prices from Oracle API...")
products = fetch_all_products(currency=DEFAULT_CURRENCY, verbose=verbose)
api_map = {p.get("partNumber", ""): p for p in products if p.get("partNumber")}
print(" {} products from API".format(len(api_map)))
family_prices = build_compute_family_lookup(api_map)
if verbose:
print("\n Family lookup built:")
for fam, prices in sorted(family_prices.items()):
print(" {:<14} OCPU=${:.4f} Memory=${:.4f}".format(
fam, prices.get("ocpu", 0), prices.get("memory", 0)
))
print("\nLoading {}...".format(COMPUTE_DOMAIN_PATH))
frontmatter, body = _load_multidoc_yaml(COMPUTE_DOMAIN_PATH)
updated = []
skipped = []
def update_shape(section_name, shape_name, shape_data, has_memory=True):
family = _shape_family(shape_name)
if not family or family not in family_prices:
skipped.append((section_name, shape_name, "no API match"))
return
prices = family_prices[family]
new_ocpu = prices.get("ocpu")
new_memory = prices.get("memory")
if new_ocpu is None:
skipped.append((section_name, shape_name, "no OCPU price"))
return
old_ocpu = shape_data.get("ocpu_per_hour") or 0
old_memory = shape_data.get("memory_per_gb_hour") or 0
# Protect against API returning $0 for tiered/free-tier SKUs (e.g. A1 Always Free).
# Same convention as refresh_catalog(): keep the existing value if it's > 0.
if new_ocpu == 0 and old_ocpu > 0:
skipped.append((section_name, shape_name, "API returned $0 (free tier) — kept existing"))
return
if has_memory and new_memory == 0 and old_memory > 0:
skipped.append((section_name, shape_name, "API returned $0 memory (free tier) — kept existing"))
return
shape_data["ocpu_per_hour"] = new_ocpu
if has_memory and new_memory is not None:
shape_data["memory_per_gb_hour"] = new_memory
# Recompute monthly_730h block if present
if "monthly_730h" in shape_data:
shape_data["monthly_730h"] = {
"ocpu": round(new_ocpu * 730, 2),
"memory_per_gb": round(new_memory * 730, 2),
}
updated.append((section_name, shape_name, family, old_ocpu, new_ocpu, old_memory, new_memory))
else:
updated.append((section_name, shape_name, family, old_ocpu, new_ocpu, None, None))
for shape_name, shape_data in (body.get("flexible_shapes") or {}).items():
if isinstance(shape_data, dict):
update_shape("flexible_shapes", shape_name, shape_data, has_memory=True)
for shape_name, shape_data in (body.get("bare_metal_shapes") or {}).items():
if isinstance(shape_data, dict):
update_shape("bare_metal_shapes", shape_name, shape_data, has_memory=False)
# Update frontmatter: bump last_verified, replace stale source, drop NEEDS REVIEW
# (the comment block lives only in the source text and is dropped on round-trip)
frontmatter["last_verified"] = date.today().isoformat()
frontmatter["source"] = API_BASE
frontmatter["description"] = (
"OCI Compute pricing for estimation purposes. Auto-refreshed from the "
"Oracle public pricing API by tools/refresh_sku_catalog.py --refresh-domain compute. "
"GPU shapes, secure desktops, estimation_helpers, and discounts are NOT auto-refreshed."
)
# Recompute estimation_helpers from the new E5 / A1 prices, since the
# hardcoded `monthly` values are derived from them.
_recompute_compute_estimation_helpers(body)
_save_multidoc_yaml(COMPUTE_DOMAIN_PATH, frontmatter, body)
print("\nRefresh complete:")
print(" Shapes updated: {}".format(len(updated)))
print(" Shapes skipped: {}".format(len(skipped)))
if verbose or updated:
print("\nUpdated shapes:")
for section, name, family, old_o, new_o, old_m, new_m in updated:
mem = ""
if old_m is not None:
mem = " mem ${:.4f}->${:.4f}".format(old_m, new_m)
print(" [{}] {} ({}): OCPU ${:.4f}->${:.4f}{}".format(
section, name, family, old_o or 0, new_o, mem
))
if skipped:
print("\nSkipped shapes (no API match — preserve existing values):")
for section, name, reason in skipped:
print(" [{}] {} ({})".format(section, name, reason))
print("\nFile saved: {}".format(COMPUTE_DOMAIN_PATH))
def _recompute_compute_estimation_helpers(body):
"""Recompute the hardcoded `monthly` values in estimation_helpers from
the (now fresh) E5 and A1 prices. Best-effort: if the referenced shape
is missing, leave the helper unchanged.
"""
helpers = body.get("estimation_helpers") or {}
flex = body.get("flexible_shapes") or {}
# Configs are: (helper_key, shape_name, ocpu_count, memory_gb)
rules = [
("typical_app_server", "VM.Standard.E5.Flex", 4, 64),
("typical_web_server", "VM.Standard.E5.Flex", 2, 16),
("typical_bastion", "VM.Standard.E5.Flex", 1, 8),
("typical_arm_app_server", "VM.Standard.A1.Flex", 4, 24),
]
for helper_key, shape_name, ocpus, mem_gb in rules:
if helper_key not in helpers:
continue
shape = flex.get(shape_name)
if not isinstance(shape, dict):
continue
ocpu_h = shape.get("ocpu_per_hour")
mem_h = shape.get("memory_per_gb_hour")
if ocpu_h is None or mem_h is None:
continue
# 730 hours per month convention (matches the rest of the file)
monthly = round((ocpus * ocpu_h * 730) + (mem_gb * mem_h * 730), 2)
helpers[helper_key]["monthly"] = monthly
def refresh_domain(domain_key, verbose=False):
"""Dispatch to the registered domain refresher."""
if domain_key not in DOMAIN_REGISTRY:
print("Unknown domain: {}. Available: {}".format(
domain_key, ", ".join(sorted(DOMAIN_REGISTRY.keys()))
))
return 1
refresher_name = DOMAIN_REGISTRY[domain_key]["refresher"]
refresher = globals()[refresher_name]
refresher(verbose=verbose)
return 0
def dump_all(output_path, verbose=False):
"""Dump raw API response to JSON file."""
products = fetch_all_products(currency=DEFAULT_CURRENCY, verbose=verbose)
@@ -307,6 +565,10 @@ def main():
)
parser.add_argument("--refresh", action="store_true",
help="Refresh catalog prices from Oracle API")
parser.add_argument("--refresh-domain", type=str, metavar="DOMAIN",
choices=sorted(DOMAIN_REGISTRY.keys()),
help="Refresh a domain pricing file from the API "
"(currently: compute)")
parser.add_argument("--validate", action="store_true",
help="Validate catalog prices against API")
parser.add_argument("--sku", type=str,
@@ -321,6 +583,8 @@ def main():
if args.sku:
inspect_sku(args.sku)
elif args.refresh_domain:
return refresh_domain(args.refresh_domain, verbose=args.verbose or args.diff)
elif args.refresh:
refresh_catalog(verbose=args.verbose or args.diff)
elif args.validate: