Company Os

#!/usr/bin/env python3
"""
Meeting Pulse Designer - Design and validate company meeting rhythms.

Generates meeting cadence recommendations, validates meeting load,
identifies redundancies, and produces L10 agenda templates.
"""

import argparse
import json
import sys
from datetime import datetime


RECOMMENDED_PULSE = {
    "daily_standup": {"frequency": "Daily", "duration_min": 15, "who": "Each team", "purpose": "Blockers only", "required": True},
    "l10_leadership": {"frequency": "Weekly", "duration_min": 90, "who": "Leadership team", "purpose": "Scorecard + IDS", "required": True},
    "department_review": {"frequency": "Monthly", "duration_min": 60, "who": "Dept + leadership", "purpose": "Deep dive", "required": True},
    "quarterly_planning": {"frequency": "Quarterly", "duration_min": 480, "who": "Leadership", "purpose": "Set rocks, review strategy", "required": True},
    "annual_planning": {"frequency": "Annual", "duration_min": 960, "who": "Leadership", "purpose": "1-year + 3-year vision", "required": True},
    "all_hands": {"frequency": "Monthly", "duration_min": 60, "who": "All employees", "purpose": "Company update + Q&A", "required": False},
    "one_on_ones": {"frequency": "Weekly", "duration_min": 30, "who": "Manager + report", "purpose": "Coaching + blockers", "required": True},
}

L10_AGENDA = [
    {"segment": "Good news", "duration_min": 5, "activity": "Personal + business wins"},
    {"segment": "Scorecard review", "duration_min": 5, "activity": "Flag red items only"},
    {"segment": "Rock review", "duration_min": 5, "activity": "On/off track for each rock"},
    {"segment": "Customer/employee headlines", "duration_min": 5, "activity": "Notable events"},
    {"segment": "Issues list (IDS)", "duration_min": 60, "activity": "Identify, Discuss, Solve"},
    {"segment": "To-dos review", "duration_min": 5, "activity": "Last week done or not?"},
    {"segment": "Conclude", "duration_min": 5, "activity": "Rate 1-10, improve next time"},
]


def design_pulse(data: dict) -> dict:
    """Design and validate meeting pulse."""
    current_meetings = data.get("current_meetings", [])
    team_size = data.get("team_size", 20)
    leadership_count = data.get("leadership_team_size", 6)

    results = {
        "timestamp": datetime.now().isoformat(),
        "team_size": team_size,
        "recommended_pulse": [],
        "current_analysis": [],
        "l10_agenda": L10_AGENDA,
        "meeting_load": {},
        "redundancies": [],
        "gaps": [],
        "recommendations": [],
    }

    # Recommended pulse
    for meeting_id, config in RECOMMENDED_PULSE.items():
        results["recommended_pulse"].append({
            "meeting": meeting_id.replace("_", " ").title(),
            "frequency": config["frequency"],
            "duration": f"{config['duration_min']} min",
            "who": config["who"],
            "purpose": config["purpose"],
            "required": config["required"],
        })

    # Analyze current meetings
    total_weekly_hours = 0
    meeting_purposes = set()

    for meeting in current_meetings:
        name = meeting.get("name", "")
        freq = meeting.get("frequency", "weekly")
        duration = meeting.get("duration_min", 60)
        attendees = meeting.get("attendees", 0)
        purpose = meeting.get("purpose", "")
        has_agenda = meeting.get("has_agenda", False)
        produces_decisions = meeting.get("produces_decisions", False)

        # Weekly equivalent hours
        freq_multiplier = {"daily": 5, "weekly": 1, "biweekly": 0.5, "monthly": 0.25, "quarterly": 0.02}
        weekly_hours = (duration / 60) * freq_multiplier.get(freq, 1) * attendees

        meeting_purposes.add(purpose.lower())

        analysis = {
            "name": name,
            "frequency": freq,
            "duration_min": duration,
            "attendees": attendees,
            "weekly_person_hours": round(weekly_hours, 1),
            "has_agenda": has_agenda,
            "produces_decisions": produces_decisions,
            "purpose": purpose,
            "issues": [],
        }

        if not has_agenda:
            analysis["issues"].append("No agenda - meetings without agendas waste time")
        if not produces_decisions and duration > 30:
            analysis["issues"].append("No decisions produced - consider if this meeting is needed")
        if duration > 90 and freq in ["daily", "weekly"]:
            analysis["issues"].append("Over 90 min for recurring meeting - energy drops after 90")

        results["current_analysis"].append(analysis)
        total_weekly_hours += weekly_hours

    # Meeting load
    hours_per_person = total_weekly_hours / team_size if team_size > 0 else 0
    results["meeting_load"] = {
        "total_weekly_person_hours": round(total_weekly_hours, 1),
        "avg_hours_per_person": round(hours_per_person, 1),
        "meeting_pct_of_week": round(hours_per_person / 40 * 100, 1),
        "assessment": "Healthy" if hours_per_person <= 8 else "Heavy" if hours_per_person <= 15 else "Excessive",
    }

    # Find redundancies
    purpose_groups = {}
    for meeting in current_meetings:
        p = meeting.get("purpose", "").lower()
        if p not in purpose_groups:
            purpose_groups[p] = []
        purpose_groups[p].append(meeting.get("name", ""))
    for purpose, meetings in purpose_groups.items():
        if len(meetings) > 1 and purpose:
            results["redundancies"].append({
                "purpose": purpose,
                "meetings": meetings,
                "recommendation": "Consolidate into one meeting or differentiate purposes",
            })

    # Find gaps
    essential = {"scorecard": False, "ids": False, "rocks": False, "standup": False, "planning": False}
    for meeting in current_meetings:
        purpose = meeting.get("purpose", "").lower()
        if "scorecard" in purpose or "metrics" in purpose:
            essential["scorecard"] = True
        if "ids" in purpose or "issues" in purpose or "problem" in purpose:
            essential["ids"] = True
        if "rock" in purpose or "quarterly" in purpose:
            essential["rocks"] = True
        if "standup" in purpose or "daily" in purpose:
            essential["standup"] = True
        if "planning" in purpose or "strategy" in purpose:
            essential["planning"] = True

    for meeting_type, present in essential.items():
        if not present:
            results["gaps"].append({
                "missing": meeting_type,
                "impact": f"No {meeting_type} meeting detected. This is a core OS component.",
            })

    # Recommendations
    if results["meeting_load"]["assessment"] == "Excessive":
        results["recommendations"].append("Meeting load is excessive. Audit and cut meetings that don't produce decisions.")
    if results["redundancies"]:
        results["recommendations"].append(f"{len(results['redundancies'])} redundant meeting group(s). Consolidate.")
    if results["gaps"]:
        results["recommendations"].append(f"{len(results['gaps'])} missing meeting type(s). Implement to complete your OS.")
    no_agenda = [m for m in results["current_analysis"] if not m["has_agenda"]]
    if no_agenda:
        results["recommendations"].append(f"{len(no_agenda)} meeting(s) without agendas. Add agendas or cancel.")

    return results


def format_text(results: dict) -> str:
    """Format as human-readable report."""
    ml = results["meeting_load"]
    lines = [
        "=" * 65,
        "MEETING PULSE DESIGN",
        "=" * 65,
        f"Team Size: {results['team_size']}  |  Meeting Load: {ml['avg_hours_per_person']:.1f} hrs/person/week ({ml['assessment']})",
        "",
        "RECOMMENDED PULSE:",
        f"{'Meeting':<25} {'Freq':<12} {'Duration':<10} {'Who':<18}",
        "-" * 65,
    ]

    for m in results["recommended_pulse"]:
        req = "*" if m["required"] else " "
        lines.append(f"{req}{m['meeting']:<24} {m['frequency']:<12} {m['duration']:<10} {m['who']:<18}")

    if results["current_analysis"]:
        lines.extend(["", "CURRENT MEETINGS:"])
        for m in results["current_analysis"]:
            issues_flag = " [!]" if m["issues"] else ""
            lines.append(f"  {m['name']:<25} {m['frequency']:<10} {m['duration_min']:>3}min {m['weekly_person_hours']:>5.1f}hrs/wk{issues_flag}")
            for issue in m["issues"]:
                lines.append(f"    -> {issue}")

    if results["gaps"]:
        lines.extend(["", "GAPS:"])
        for g in results["gaps"]:
            lines.append(f"  [!] Missing: {g['missing']} - {g['impact']}")

    if results["redundancies"]:
        lines.extend(["", "REDUNDANCIES:"])
        for r in results["redundancies"]:
            lines.append(f"  [!] '{r['purpose']}' covered by: {', '.join(r['meetings'])}")

    lines.extend(["", "L10 MEETING AGENDA (90 min):"])
    for segment in L10_AGENDA:
        lines.append(f"  {segment['duration_min']:>2} min | {segment['segment']:<30} | {segment['activity']}")

    if results["recommendations"]:
        lines.extend(["", "RECOMMENDATIONS:"])
        for r in results["recommendations"]:
            lines.append(f"  -> {r}")

    lines.extend(["", "* = required for effective OS", "=" * 65])
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Design and validate meeting pulse")
    parser.add_argument("--input", "-i", help="JSON file with meeting data")
    parser.add_argument("--team-size", type=int, default=35, help="Company team size")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    if args.input:
        with open(args.input) as f:
            data = json.load(f)
    else:
        data = {
            "team_size": args.team_size,
            "leadership_team_size": 6,
            "current_meetings": [
                {"name": "Engineering standup", "frequency": "daily", "duration_min": 15, "attendees": 12, "purpose": "daily standup", "has_agenda": True, "produces_decisions": False},
                {"name": "Leadership sync", "frequency": "weekly", "duration_min": 60, "attendees": 6, "purpose": "scorecard and updates", "has_agenda": True, "produces_decisions": True},
                {"name": "Product review", "frequency": "weekly", "duration_min": 60, "attendees": 8, "purpose": "product updates", "has_agenda": True, "produces_decisions": True},
                {"name": "Sales pipeline", "frequency": "weekly", "duration_min": 45, "attendees": 5, "purpose": "pipeline review", "has_agenda": False, "produces_decisions": False},
                {"name": "All hands", "frequency": "monthly", "duration_min": 60, "attendees": 35, "purpose": "company update", "has_agenda": True, "produces_decisions": False},
                {"name": "Status update", "frequency": "weekly", "duration_min": 30, "attendees": 6, "purpose": "scorecard and updates", "has_agenda": False, "produces_decisions": False},
            ],
        }

    results = design_pulse(data)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_text(results))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
90-Day Rocks Tracker - Track quarterly rocks with binary status and accountability.

Manages company and individual rocks with on/off track status,
accountability assignments, and quarterly review preparation.
"""

import argparse
import json
import sys
from datetime import datetime


def track_rocks(data: dict) -> dict:
    """Track 90-day rocks status."""
    company_rocks = data.get("company_rocks", [])
    individual_rocks = data.get("individual_rocks", [])
    quarter = data.get("quarter", "Q1")

    results = {
        "timestamp": datetime.now().isoformat(),
        "quarter": quarter,
        "company_rocks": {"total": 0, "on_track": 0, "off_track": 0, "complete": 0, "rocks": []},
        "individual_summary": {},
        "individual_rocks": [],
        "issues": [],
        "recommendations": [],
    }

    for rock in company_rocks:
        status = rock.get("status", "on_track")
        result = {
            "description": rock.get("description", ""),
            "owner": rock.get("owner", ""),
            "status": status,
            "due_date": rock.get("due_date", ""),
            "progress_note": rock.get("progress_note", ""),
            "blocker": rock.get("blocker", ""),
        }
        results["company_rocks"]["rocks"].append(result)
        results["company_rocks"]["total"] += 1
        if status == "complete":
            results["company_rocks"]["complete"] += 1
        elif status == "on_track":
            results["company_rocks"]["on_track"] += 1
        else:
            results["company_rocks"]["off_track"] += 1
            if result["blocker"]:
                results["issues"].append({
                    "rock": result["description"][:50],
                    "owner": result["owner"],
                    "blocker": result["blocker"],
                    "type": "company_rock",
                })

    for rock in individual_rocks:
        owner = rock.get("owner", "")
        status = rock.get("status", "on_track")
        result = {
            "description": rock.get("description", ""),
            "owner": owner,
            "status": status,
            "due_date": rock.get("due_date", ""),
        }
        results["individual_rocks"].append(result)

        if owner not in results["individual_summary"]:
            results["individual_summary"][owner] = {"total": 0, "on_track": 0, "off_track": 0, "complete": 0}
        results["individual_summary"][owner]["total"] += 1
        if status == "complete":
            results["individual_summary"][owner]["complete"] += 1
        elif status == "on_track":
            results["individual_summary"][owner]["on_track"] += 1
        else:
            results["individual_summary"][owner]["off_track"] += 1

    # Recommendations
    cr = results["company_rocks"]
    if cr["total"] > 7:
        results["recommendations"].append(f"Too many company rocks ({cr['total']}). Max 7. Prioritize ruthlessly.")
    if cr["off_track"] > 0:
        results["recommendations"].append(f"{cr['off_track']} company rock(s) off track. Address in L10.")

    for owner, summary in results["individual_summary"].items():
        if summary["total"] > 7:
            results["recommendations"].append(f"{owner} has {summary['total']} rocks (max 7). Reduce or delegate.")
        if summary["off_track"] >= 2:
            results["recommendations"].append(f"{owner} has {summary['off_track']} rocks off track. Capacity issue.")

    return results


def format_text(results: dict) -> str:
    """Format as L10-ready rocks review."""
    cr = results["company_rocks"]
    lines = [
        "=" * 65,
        f"90-DAY ROCKS TRACKER - {results['quarter']}",
        "=" * 65,
        f"Company: {cr['complete']} done / {cr['on_track']} on track / {cr['off_track']} off track (of {cr['total']})",
        "",
        "COMPANY ROCKS:",
    ]

    for r in cr["rocks"]:
        icon = "[x]" if r["status"] == "complete" else "[~]" if r["status"] == "on_track" else "[!]"
        lines.append(f"  {icon} {r['description']} ({r['owner']}, due: {r['due_date']})")
        if r["blocker"]:
            lines.append(f"      Blocker: {r['blocker']}")

    lines.extend(["", "INDIVIDUAL ROCKS BY OWNER:"])
    for owner, summary in results["individual_summary"].items():
        lines.append(f"  {owner}: {summary['complete']}x / {summary['on_track']}~ / {summary['off_track']}! (of {summary['total']})")

    if results["issues"]:
        lines.extend(["", "ISSUES FOR IDS:"])
        for issue in results["issues"]:
            lines.append(f"  >> {issue['rock']}... ({issue['owner']}): {issue['blocker']}")

    if results["recommendations"]:
        lines.extend(["", "RECOMMENDATIONS:"])
        for r in results["recommendations"]:
            lines.append(f"  -> {r}")

    lines.extend(["", "=" * 65])
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Track 90-day rocks")
    parser.add_argument("--input", "-i", help="JSON file with rocks data")
    parser.add_argument("--quarter", default="Q1", help="Quarter label")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    if args.input:
        with open(args.input) as f:
            data = json.load(f)
    else:
        data = {
            "quarter": args.quarter,
            "company_rocks": [
                {"description": "Hit $3M ARR", "owner": "CRO", "status": "on_track", "due_date": "2026-03-31"},
                {"description": "Launch enterprise tier", "owner": "CPO", "status": "off_track", "due_date": "2026-03-31", "blocker": "API redesign delayed by 3 weeks"},
                {"description": "Achieve SOC 2 Type I", "owner": "CISO", "status": "on_track", "due_date": "2026-03-31"},
                {"description": "Hire 5 key roles", "owner": "CHRO", "status": "off_track", "due_date": "2026-03-31", "blocker": "2 offers declined, comp bands may need refresh"},
            ],
            "individual_rocks": [
                {"description": "Close 3 enterprise deals", "owner": "CRO", "status": "on_track", "due_date": "2026-03-31"},
                {"description": "Reduce MTTR to < 4 hours", "owner": "CTO", "status": "complete", "due_date": "2026-03-31"},
                {"description": "Launch content hub", "owner": "CMO", "status": "on_track", "due_date": "2026-03-31"},
                {"description": "Implement weekly L10", "owner": "COO", "status": "complete", "due_date": "2026-02-28"},
                {"description": "Build hiring scorecard v2", "owner": "CHRO", "status": "off_track", "due_date": "2026-03-31"},
            ],
        }

    results = track_rocks(data)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_text(results))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Company Scorecard Builder - Build and track weekly company scorecards.

Manages 5-15 weekly metrics with owners, targets, and RAG status.
Identifies trends, flags issues for L10 discussion, and generates
IDS-ready issue lists.
"""

import argparse
import json
import sys
from datetime import datetime


def build_scorecard(data: dict) -> dict:
    """Build company scorecard with status and trends."""
    metrics = data.get("metrics", [])
    weeks_of_data = data.get("weeks", 1)

    results = {
        "timestamp": datetime.now().isoformat(),
        "metrics_count": len(metrics),
        "summary": {"green": 0, "yellow": 0, "red": 0},
        "scorecard": [],
        "issues_for_ids": [],
        "rocks_status": [],
        "recommendations": [],
    }

    if len(metrics) > 15:
        results["recommendations"].append(f"Too many metrics ({len(metrics)}). Maximum 15. Cut to focus attention.")
    elif len(metrics) < 5:
        results["recommendations"].append(f"Only {len(metrics)} metrics. Consider adding to cover blind spots.")

    for m in metrics:
        name = m.get("name", "")
        owner = m.get("owner", "Unassigned")
        target = m.get("target", 0)
        current = m.get("current", 0)
        previous = m.get("previous", None)
        direction = m.get("direction", "higher_is_better")
        fmt = m.get("format", "number")
        weeks_red = m.get("consecutive_weeks_red", 0)

        # RAG calculation
        if target == 0:
            status = "N/A"
        elif direction == "higher_is_better":
            ratio = current / target
            status = "G" if ratio >= 0.95 else "Y" if ratio >= 0.85 else "R"
        elif direction == "lower_is_better":
            ratio = current / target
            status = "G" if ratio <= 1.05 else "Y" if ratio <= 1.15 else "R"
        else:
            status = "G" if abs(current - target) / target < 0.05 else "Y" if abs(current - target) / target < 0.10 else "R"

        # Trend
        if previous is not None:
            if direction == "higher_is_better":
                trend = "Improving" if current > previous else "Declining" if current < previous else "Flat"
            else:
                trend = "Improving" if current < previous else "Declining" if current > previous else "Flat"
        else:
            trend = "N/A"

        metric_result = {
            "name": name,
            "owner": owner,
            "target": target,
            "current": current,
            "previous": previous,
            "status": status,
            "trend": trend,
            "consecutive_weeks_red": weeks_red + (1 if status == "R" else 0),
            "needs_discussion": status == "R",
        }
        results["scorecard"].append(metric_result)

        if status == "G":
            results["summary"]["green"] += 1
        elif status == "Y":
            results["summary"]["yellow"] += 1
        elif status == "R":
            results["summary"]["red"] += 1
            results["issues_for_ids"].append({
                "metric": name,
                "owner": owner,
                "current_value": current,
                "target": target,
                "weeks_red": metric_result["consecutive_weeks_red"],
                "suggested_ids_format": f"IDENTIFY: {name} is below target ({current} vs {target}). Root cause?",
            })

    # Rocks status
    rocks = data.get("rocks", [])
    for rock in rocks:
        results["rocks_status"].append({
            "rock": rock.get("description", ""),
            "owner": rock.get("owner", ""),
            "status": rock.get("status", "on_track"),
            "due": rock.get("due_date", ""),
        })

    # Recommendations
    red_count = results["summary"]["red"]
    if red_count >= 3:
        results["recommendations"].append(f"{red_count} metrics in RED. Prioritize in L10 IDS discussion.")
    chronic_red = [m for m in results["scorecard"] if m["consecutive_weeks_red"] >= 3]
    if chronic_red:
        results["recommendations"].append(
            f"{len(chronic_red)} metric(s) red for 3+ weeks: {', '.join(m['name'] for m in chronic_red)}. "
            "Either the target is wrong or the plan isn't working. Force resolution."
        )

    off_track_rocks = [r for r in results["rocks_status"] if r["status"] == "off_track"]
    if off_track_rocks:
        results["recommendations"].append(f"{len(off_track_rocks)} rock(s) off track. Address in L10.")

    return results


def format_text(results: dict) -> str:
    """Format as L10-ready scorecard."""
    s = results["summary"]
    lines = [
        "=" * 70,
        "WEEKLY SCORECARD",
        "=" * 70,
        f"Date: {results['timestamp'][:10]}  |  {s['green']}G / {s['yellow']}Y / {s['red']}R",
        "",
        f"{'Metric':<22} {'Owner':<10} {'Current':>10} {'Target':>10} {'RAG':>5} {'Trend':<12}",
        "-" * 70,
    ]

    for m in results["scorecard"]:
        rag = f"[{m['status']}]"
        discuss = " <<" if m["needs_discussion"] else ""
        lines.append(
            f"{m['name']:<22} {m['owner']:<10} {m['current']:>10} {m['target']:>10} "
            f"{rag:>5} {m['trend']:<12}{discuss}"
        )

    if results["issues_for_ids"]:
        lines.extend(["", "IDS ISSUES (discuss in L10):"])
        for issue in results["issues_for_ids"]:
            lines.append(f"  >> {issue['suggested_ids_format']} ({issue['owner']}, {issue['weeks_red']}w red)")

    if results["rocks_status"]:
        lines.extend(["", "ROCKS:"])
        for r in results["rocks_status"]:
            icon = "[x]" if r["status"] == "complete" else "[~]" if r["status"] == "on_track" else "[!]"
            lines.append(f"  {icon} {r['rock']} ({r['owner']}, due: {r['due']})")

    if results["recommendations"]:
        lines.extend(["", "RECOMMENDATIONS:"])
        for r in results["recommendations"]:
            lines.append(f"  -> {r}")

    lines.extend(["", "<< = needs L10 discussion", "=" * 70])
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Build and track weekly company scorecard")
    parser.add_argument("--input", "-i", help="JSON file with scorecard data")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    if args.input:
        with open(args.input) as f:
            data = json.load(f)
    else:
        data = {
            "metrics": [
                {"name": "New MRR", "owner": "CRO", "target": 50000, "current": 43000, "previous": 47000, "direction": "higher_is_better", "consecutive_weeks_red": 1},
                {"name": "Logo Churn", "owner": "CS Lead", "target": 1.0, "current": 0.8, "previous": 0.9, "direction": "lower_is_better", "consecutive_weeks_red": 0},
                {"name": "Active Users", "owner": "CPO", "target": 2000, "current": 2150, "previous": 2050, "direction": "higher_is_better", "consecutive_weeks_red": 0},
                {"name": "Deployments/wk", "owner": "CTO", "target": 3, "current": 3, "previous": 3, "direction": "higher_is_better", "consecutive_weeks_red": 0},
                {"name": "Critical Bugs", "owner": "CTO", "target": 0, "current": 2, "previous": 1, "direction": "lower_is_better", "consecutive_weeks_red": 2},
                {"name": "Runway (mo)", "owner": "CFO", "target": 18, "current": 16, "previous": 17, "direction": "higher_is_better", "consecutive_weeks_red": 3},
                {"name": "Offer Accept %", "owner": "CHRO", "target": 85, "current": 90, "previous": 80, "direction": "higher_is_better", "consecutive_weeks_red": 0},
            ],
            "rocks": [
                {"description": "Implement CRM with pipeline stages", "owner": "CRO", "status": "on_track", "due_date": "2026-03-31"},
                {"description": "Hire 3 senior engineers", "owner": "CTO", "status": "off_track", "due_date": "2026-04-15"},
                {"description": "Reduce churn from 3% to 1.5%", "owner": "CS Lead", "status": "on_track", "due_date": "2026-06-30"},
            ],
        }

    results = build_scorecard(data)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_text(results))


if __name__ == "__main__":
    main()