@clawhub-harrylabsj-35a31b2850
Design a habit stack by choosing a reliable existing anchor, shrinking the new habit into a tiny first action, and adding friction reduction, reward, and a r...
---
name: habit-stack-designer
description: Design a habit stack by choosing a reliable existing anchor, shrinking the new habit into a tiny first action, and adding friction reduction, reward, and a rescue version for hard days. Use when the user wants to build a habit without relying on memory, motivation spikes, or giant behavior changes.
---
# Habit Stack Designer
## Overview
Use this skill to attach a new habit to an existing routine instead of hoping motivation will carry it. It helps the user find a stable anchor, make the first action tiny enough to start, and design friction removal plus a rescue version for bad days.
This skill is descriptive only. It does not create reminders or automation.
## Trigger
Use this skill when the user wants to:
- build a new habit that actually sticks
- choose a strong anchor habit instead of relying on memory
- shrink a habit into a tiny under-two-minute version
- create an explicit habit stack formula
- add a rescue version for low-energy days
### Example prompts
- "Help me stack a meditation habit onto something I already do"
- "Design a tiny reading habit after dinner"
- "I keep forgetting my new habit, can you attach it to a stable routine?"
- "Turn this big habit goal into a stack formula"
## Workflow
1. Identify the target habit and why it matters.
2. Find an existing routine that already happens predictably.
3. Evaluate whether the anchor is frequent and reliable.
4. Shrink the new habit into a tiny first action.
5. Write the habit stack formula.
6. Add friction reduction, reward, and a rescue version.
7. Scale only after consistency exists.
## Inputs
The user can provide any mix of:
- target habit
- reason or identity behind the habit
- daily routines that already happen
- time or place cues
- schedule constraints
- prior failure patterns
- desired reward or review rhythm
## Outputs
Return a markdown habit stack design with:
- target habit and why it matters
- best anchor and why it is reliable
- explicit stack formula
- friction removal ideas
- reward loop
- rescue version for hard days
## Safety
- Prefer anchors the user already performs several times per week.
- Start tiny enough to avoid negotiation.
- Avoid stacking many new behaviors onto one weak anchor.
- Irregular schedules may need place-based anchors rather than clock-based ones.
## Acceptance Criteria
- Return markdown text.
- Include a clear anchor and explicit stack formula sentence.
- Keep the starting action tiny and realistic.
- Include both friction removal and a rescue version.
FILE:handler.py
#!/usr/bin/env python3
import json
import os
import re
import sys
from typing import Any, Dict, List, Tuple
def _load_skill_meta(slug):
path = os.path.join(os.path.dirname(__file__), 'SKILL.md')
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
meta = re.search(r'^---$(.*?)^---$', content, re.DOTALL | re.MULTILINE)
return meta.group(1).strip() if meta else ''
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ''
if isinstance(inputs, str):
return inputs.strip()
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
HABIT_RULES: List[Tuple[str, List[str]]] = [
('meditation', ['meditate', 'meditation']),
('reading', ['read', 'reading', 'book']),
('stretching', ['stretch', 'mobility', 'yoga']),
('journaling', ['journal', 'write']),
('studying', ['study', 'revision', 'homework', 'practice problem']),
('hydration', ['drink water', 'hydration', 'water']),
('walking', ['walk', 'steps']),
('tidying', ['tidy', 'declutter', 'clean up']),
('flossing', ['floss']),
]
ANCHOR_RULES: List[Tuple[str, List[str], str]] = [
('brush my teeth', ['brush my teeth', 'toothbrush', 'brushing'], 'It already happens in a fixed place and rarely gets skipped.'),
('make coffee or tea', ['coffee', 'tea', 'kettle'], 'It is a repeatable ritual that naturally creates a transition point.'),
('sit at my desk', ['open my laptop', 'sit at my desk', 'desk'], 'It happens near the exact moment focus is needed.'),
('finish lunch', ['after lunch', 'lunch'], 'It is a clear daily event with a stable sequence.'),
('take a shower', ['shower'], 'It is anchored to a physical environment and hard to forget.'),
('finish dinner', ['after dinner', 'dinner'], 'It gives the habit a predictable evening landing spot.'),
('put my phone on the charger', ['charger', 'charge my phone', 'plug in my phone'], 'It creates a strong bedtime cue with a visible object.'),
]
TINY_HABITS = {
'meditation': 'take 3 slow breaths with eyes closed',
'reading': 'read one page',
'stretching': 'do 60 seconds of stretching',
'journaling': 'write one sentence',
'studying': 'open the material and solve one small problem',
'hydration': 'take 5 sips of water',
'walking': 'walk for 2 minutes',
'tidying': 'put away 3 items',
'flossing': 'floss one tooth',
}
RESCUE_HABITS = {
'meditation': 'take one slow breath and call it a win.',
'reading': 'read one sentence and stop if needed.',
'stretching': 'do one stretch for 10 seconds.',
'journaling': 'write three words about the day.',
'studying': 'open the book or file and look at one question.',
'hydration': 'take one sip of water.',
'walking': 'stand up and take ten steps.',
'tidying': 'put away one item.',
'flossing': 'floss one tooth.',
}
WHY_TEMPLATES = {
'meditation': 'It helps you create a calmer start and better attention.',
'reading': 'It helps you stay in contact with ideas instead of waiting for a perfect long block.',
'stretching': 'It reduces stiffness and makes movement easier to begin.',
'journaling': 'It helps you clear mental clutter and notice your own patterns.',
'studying': 'It protects consistent learning over last-minute cramming.',
'hydration': 'It supports energy and makes the habit almost frictionless.',
'walking': 'It builds movement into ordinary life without needing a full workout.',
'tidying': 'It lowers visual friction and keeps the space easier to use.',
'flossing': 'It makes oral care automatic instead of optional.',
}
class HabitStackDesigner:
def __init__(self, inputs: Any):
self.raw = _normalize_inputs(inputs)
self.lower = self.raw.lower()
self.target_habit = self._extract_target_habit()
self.why = self._extract_why()
self.anchor, self.anchor_reason = self._extract_anchor()
self.tiny_habit = TINY_HABITS[self.target_habit]
self.stack_formula = f'After I {self.anchor}, I will {self.tiny_habit}.'
self.friction_removal = self._friction_removal()
self.reward = self._reward_loop()
self.rescue_version = RESCUE_HABITS[self.target_habit]
def _extract_target_habit(self) -> str:
for label, keywords in HABIT_RULES:
if any(keyword in self.lower for keyword in keywords):
return label
return 'journaling'
def _extract_why(self) -> str:
match = re.search(r'(?:because|so that)\s+(.+?)(?:[.?!]|$)', self.raw, flags=re.IGNORECASE)
if match:
return match.group(1).strip()
return WHY_TEMPLATES[self.target_habit]
def _extract_anchor(self) -> Tuple[str, str]:
for label, keywords, reason in ANCHOR_RULES:
if any(keyword in self.lower for keyword in keywords):
if any(word in self.lower for word in ['shift', 'rotating', 'irregular']):
reason = reason + ' It is also place-based, which helps when the schedule changes.'
return label, reason
return 'brush my teeth', 'It already happens daily and gives the new habit a strong physical cue.'
def _friction_removal(self) -> str:
if self.target_habit == 'reading':
return 'Keep the book where the anchor happens and stop choosing the next page in advance.'
if self.target_habit == 'meditation':
return 'Put a cushion, chair, or visual cue exactly where the anchor ends.'
if self.target_habit == 'stretching':
return 'Lay out a mat or choose one default stretch so there is no decision cost.'
if self.target_habit == 'journaling':
return 'Keep the notebook open with a pen ready at the anchor location.'
return 'Prepare the smallest required tool before the anchor happens so the habit starts in one motion.'
def _reward_loop(self) -> Dict[str, str]:
return {
'immediate': 'Mark one check, say "done," and enjoy the quick feeling of completion before moving on.',
'weekly': 'Review the stack once at the end of the week and only scale it if it felt easy to repeat.',
}
def render(self) -> str:
lines: List[str] = []
lines.append('# Habit Stack Design')
lines.append('')
lines.append('## New Habit')
lines.append(f'- Target habit: {self.target_habit}')
lines.append(f'- Why it matters: {self.why}')
lines.append('')
lines.append('## Best Anchor')
lines.append(f'- Existing routine: {self.anchor}')
lines.append(f'- Why it is reliable: {self.anchor_reason}')
lines.append('')
lines.append('## Stack Formula')
lines.append(f'- {self.stack_formula}')
lines.append('')
lines.append('## Friction Removal')
lines.append(f'- {self.friction_removal}')
lines.append('')
lines.append('## Reward Loop')
lines.append(f"- Immediate reward: {self.reward['immediate']}")
lines.append(f"- Weekly review cue: {self.reward['weekly']}")
lines.append('')
lines.append('## Rescue Version')
lines.append(f'- Minimum version for hard days: {self.rescue_version}')
return '\n'.join(lines)
def handle(inputs):
_load_skill_meta('habit-stack-designer')
designer = HabitStackDesigner(inputs)
return designer.render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import HabitStackDesigner, handle
def test_target_habit_detection():
designer = HabitStackDesigner('I want to meditate every day because I need a calmer start.')
assert designer.target_habit == 'meditation'
def test_anchor_detection():
designer = HabitStackDesigner('I want to read more after I brush my teeth each night.')
assert designer.anchor == 'brush my teeth'
def test_output_sections():
output = handle('I want to meditate after I brush my teeth because I need a calmer start.')
assert output.startswith('# Habit Stack Design')
assert 'After I brush my teeth, I will' in output
assert '## Rescue Version' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Turn freeform notes into a specific gratitude entry with concrete moments, why they mattered, the feeling they created, and a small follow-up action when use...
---
name: gratitude-journalist
description: Turn freeform notes into a specific gratitude entry with concrete moments, why they mattered, the feeling they created, and a small follow-up action when useful. Use when the user wants a gratitude practice that feels honest, vivid, and varied instead of repetitive, forced, or emotionally flat.
---
# Gratitude Journalist
## Overview
Use this skill to turn gratitude from a generic list into a vivid record of specific moments. It helps the user notice people, body, environment, luck, learning, and self-kindness without forcing fake positivity or ignoring hard days.
This skill is descriptive only. It does not post, message, or track mood data.
## Trigger
Use this skill when the user wants to:
- write a gratitude entry that feels specific and alive
- avoid repeating the same generic gratitude phrases
- reflect on ordinary moments, not only major wins
- add emotional meaning and a small follow-up action to gratitude notes
- keep gratitude honest during a difficult day or week
### Example prompts
- "Help me write a gratitude entry that does not feel cheesy"
- "Turn these notes into a gratitude journal entry"
- "Give me 3 specific gratitude moments from today"
- "I want a gratitude practice that still feels honest on hard days"
## Workflow
1. Capture two to five gratitude moments from today or this week.
2. Push toward concrete scenes instead of vague categories.
3. Rotate the lens across people, body, environment, luck, learning, and self-kindness.
4. For each item, ask why it mattered and what feeling it created.
5. Suggest a micro follow-up action when useful.
6. End with one pattern to notice.
## Inputs
The user can provide any mix of:
- rough gratitude notes
- moments from today or this week
- relationship moments
- body or health moments
- environment or weather details
- small luck, support, or learning moments
- a note that the day felt hard, heavy, or exhausting
## Outputs
Return a markdown gratitude entry with:
- two to five specific gratitude items
- why each item mattered
- the feeling it created
- follow-up action when relevant
- one pattern to notice
## Safety
- Do not pressure the user to feel grateful for harmful situations.
- On painful days, let small relief count without invalidating grief, anger, or exhaustion.
- Prefer emotionally honest wording over polished positivity.
- Do not claim gratitude alone will solve serious distress.
## Acceptance Criteria
- Return markdown text.
- Include specific, pictureable moments.
- Vary the lens when the input is repetitive.
- End with a pattern to notice.
FILE:handler.py
#!/usr/bin/env python3
import json
import os
import re
import sys
from typing import Any, Dict, List
def _load_skill_meta(slug):
path = os.path.join(os.path.dirname(__file__), 'SKILL.md')
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
meta = re.search(r'^---$(.*?)^---$', content, re.DOTALL | re.MULTILINE)
return meta.group(1).strip() if meta else ''
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ''
if isinstance(inputs, str):
return inputs.strip()
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
LENS_KEYWORDS = {
'people': ['friend', 'partner', 'family', 'child', 'kid', 'parent', 'teacher', 'neighbor', 'someone'],
'body': ['body', 'health', 'sleep', 'walk', 'breath', 'rest', 'energy', 'exercise'],
'environment': ['sun', 'rain', 'home', 'window', 'quiet', 'tea', 'coffee', 'tree', 'weather', 'room'],
'luck': ['lucky', 'unexpected', 'chance', 'timing', 'helped', 'worked out'],
'learning': ['learn', 'book', 'read', 'lesson', 'insight', 'practice'],
'self-kindness': ['rest', 'pause', 'gentle', 'forgave', 'kind to myself', 'slow down'],
}
DEFAULT_LENS_ORDER = ['people', 'body', 'environment', 'learning', 'self-kindness']
WHY_BY_LENS = {
'people': 'it reminded me that support often arrives through ordinary gestures.',
'body': 'it showed me that my body is still helping me, even in small ways.',
'environment': 'it gave the day texture, steadiness, or beauty instead of pure rush.',
'luck': 'it lowered friction at the right moment and made the day feel a little lighter.',
'learning': 'it gave me something useful to carry forward, not just consume and forget.',
'self-kindness': 'it interrupted self-pressure and created a little more room to breathe.',
}
FEELING_BY_LENS = {
'people': 'Warmth and connection.',
'body': 'Relief and appreciation.',
'environment': 'Calm and steadiness.',
'luck': 'Surprise and gratitude.',
'learning': 'Hope and grounded confidence.',
'self-kindness': 'Gentleness and self-respect.',
}
FOLLOW_UP_BY_LENS = {
'people': 'Send a short thank-you note or mention it next time you talk.',
'body': 'Protect one tiny behavior that supported this moment.',
'environment': 'Notice how to recreate this setting tomorrow.',
'luck': 'Write down what made the good timing possible, if anything.',
'learning': 'Capture the lesson in one line so it sticks.',
'self-kindness': 'Repeat the same gentle move when stress rises again.',
}
FALLBACK_MOMENTS = {
'people': 'a small moment of support from someone close to me',
'body': 'one sign my body kept carrying me today',
'environment': 'a calm detail in my environment that softened the day',
'learning': 'one thing I learned that made me a little wiser or steadier',
'self-kindness': 'one moment I treated myself with less pressure than usual',
}
class GratitudeJournalist:
def __init__(self, inputs: Any):
self.raw = _normalize_inputs(inputs)
self.lower = self.raw.lower()
self.hard_day = any(word in self.lower for word in ['hard', 'heavy', 'grief', 'angry', 'tired', 'exhausted', 'rough'])
self.entries = self._build_entries()
self.pattern = self._detect_pattern()
def _chunks(self) -> List[str]:
text = self.raw.replace('\n', ';')
pieces = re.split(r'[;•]+', text)
if len(pieces) <= 1:
pieces = re.split(r'(?<=[.!?])\s+', text)
cleaned = []
for piece in pieces:
piece = re.sub(r'^\s*(\d+[.)-]?|[-*])\s*', '', piece).strip()
if piece:
cleaned.append(piece)
return cleaned[:5]
def _infer_lens(self, chunk: str) -> str:
lower = chunk.lower()
for lens, keywords in LENS_KEYWORDS.items():
if any(keyword in lower for keyword in keywords):
return lens
return 'environment'
def _fallback_entries(self) -> List[Dict[str, str]]:
entries: List[Dict[str, str]] = []
for lens in DEFAULT_LENS_ORDER[:3]:
entries.append({
'lens': lens,
'item': FALLBACK_MOMENTS[lens],
'why': WHY_BY_LENS[lens],
'feeling': 'Small relief and honesty.' if self.hard_day else FEELING_BY_LENS[lens],
'follow_up': FOLLOW_UP_BY_LENS[lens],
})
return entries
def _build_entries(self) -> List[Dict[str, str]]:
chunks = self._chunks()
if len(chunks) < 2:
return self._fallback_entries()
entries: List[Dict[str, str]] = []
for chunk in chunks[:5]:
lens = self._infer_lens(chunk)
item = chunk[:1].upper() + chunk[1:]
entries.append({
'lens': lens,
'item': item,
'why': WHY_BY_LENS[lens],
'feeling': 'Relief and steadiness.' if self.hard_day and lens != 'people' else FEELING_BY_LENS[lens],
'follow_up': FOLLOW_UP_BY_LENS[lens],
})
return entries
def _detect_pattern(self) -> str:
if self.hard_day:
return 'Even on a hard day, small supports and brief moments of relief still counted.'
lenses = [entry['lens'] for entry in self.entries]
if lenses.count('people') >= 2:
return 'Support is showing up through ordinary people and small gestures.'
if any(lens in lenses for lens in ['environment', 'body']):
return 'Small steady details are carrying more of the day than dramatic events.'
return 'Ordinary moments are becoming easier to notice when you slow down long enough to name them.'
def render(self) -> str:
lines: List[str] = []
lines.append('# Gratitude Entry')
lines.append('')
lines.append('## Today I am grateful for')
for index, entry in enumerate(self.entries, 1):
lines.append(f"{index}. {entry['item']}")
lines.append(f" - Why it mattered: {entry['why']}")
lines.append(f" - Feeling it created: {entry['feeling']}")
lines.append(f" - Follow-up action, if any: {entry['follow_up']}")
lines.append('')
lines.append('## Pattern to Notice')
lines.append(f'- {self.pattern}')
return '\n'.join(lines)
def handle(inputs):
_load_skill_meta('gratitude-journalist')
journal = GratitudeJournalist(inputs)
return journal.render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import GratitudeJournalist, handle
def test_entry_count_from_input():
journal = GratitudeJournalist('My friend checked in on me; the quiet walk after dinner')
assert len(journal.entries) >= 2
def test_people_lens_detection():
journal = GratitudeJournalist('My friend checked in on me; a sunny window in the afternoon')
assert journal.entries[0]['lens'] == 'people'
def test_output_structure():
output = handle('My son laughed with me; I noticed a calm cup of tea after dinner')
assert output.startswith('# Gratitude Entry')
assert output.count('- Why it mattered:') >= 2
assert '## Pattern to Notice' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Map a decision into options, non-negotiables, criteria, best/base/worst cases, reversibility, and a review checkpoint. Use when the user is comparing multipl...
---
name: decision-forest
description: Map a decision into options, non-negotiables, criteria, best/base/worst cases, reversibility, and a review checkpoint. Use when the user is comparing multiple paths and needs a clear decision aid that separates facts, assumptions, and fears without pretending to predict the future.
---
# Decision Forest
## Overview
Use this skill to slow a messy decision down enough to see what actually matters. It helps the user define the real question, compare options branch by branch, separate facts from assumptions and fears, and end with either a provisional choice or a clearly bounded next test.
This skill is descriptive only. It does not provide legal, medical, or financial advice, and it does not predict outcomes.
## Trigger
Use this skill when the user wants to:
- compare two or more life, work, or project options
- stop mixing facts, fears, and imagined outcomes together
- add reversibility and future-regret checks to a decision
- narrow a crowded option set
- choose a small test instead of forcing immediate certainty
### Example prompts
- "Help me decide whether to stay in my job or start a small consulting business"
- "Build a decision tree for moving, waiting, or testing a smaller version"
- "I am stuck between two options and I cannot see what matters most"
- "Map best case, base case, and worst case for this decision"
## Workflow
1. Define the decision in one sentence.
2. List the visible options and the hidden option of delaying or testing a smaller version.
3. Identify non-negotiables and decision criteria.
4. Build branches for each option.
5. Sketch best case, base case, and worst case outcomes.
6. Check reversibility and future regret.
7. Recommend a provisional choice and a review checkpoint.
## Inputs
The user can provide any mix of:
- the decision question
- visible options
- deadline or time pressure
- constraints and non-negotiables
- known facts or data points
- fears, hopes, or gut reactions
- practical criteria, such as cost, time, energy, meaning, family impact, or reversibility
## Outputs
Return a markdown decision brief with:
- decision trunk summary
- non-negotiables and criteria
- branch review for each option
- best, base, and worst cases
- reversibility and key unknowns
- provisional choice logic and review date
## Safety
- Keep facts, assumptions, and fears visibly separate.
- High-stakes legal, medical, or financial decisions may need expert advice beyond heuristics.
- Do not claim certainty the information cannot support.
- When uncertainty is high, prefer a smaller reversible test over forced confidence.
## Acceptance Criteria
- Return markdown text.
- Include at least three meaningful criteria.
- Show reversibility for each branch.
- End with either a provisional choice or a very clear next test.
FILE:handler.py
#!/usr/bin/env python3
import json
import os
import re
import sys
from typing import Any, Dict, List
def _load_skill_meta(slug):
path = os.path.join(os.path.dirname(__file__), 'SKILL.md')
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
meta = re.search(r'^---$(.*?)^---$', content, re.DOTALL | re.MULTILINE)
return meta.group(1).strip() if meta else ''
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ''
if isinstance(inputs, str):
return inputs.strip()
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _clean_option(text: str) -> str:
text = text.strip().strip(' ?.!')
text = re.sub(r'^(whether to|to|either)\s+', '', text, flags=re.IGNORECASE)
return text[:1].upper() + text[1:] if text else 'Unnamed option'
CRITERIA_RULES = {
'Cost and runway': ['cost', 'money', 'salary', 'income', 'budget', 'cash', 'runway'],
'Time and life load': ['time', 'busy', 'schedule', 'bandwidth', 'capacity'],
'Energy and stress': ['energy', 'stress', 'burnout', 'tired'],
'Meaning and growth': ['meaning', 'growth', 'purpose', 'learn', 'career'],
'Family impact': ['family', 'kids', 'partner', 'home'],
'Reversibility': ['reverse', 'reversible', 'test', 'trial', 'pilot'],
}
class DecisionForestPlanner:
def __init__(self, inputs: Any):
self.raw = _normalize_inputs(inputs)
self.lower = self.raw.lower()
self.snippets = self._split_snippets()
self.decision = self._extract_decision()
self.deadline = self._extract_deadline()
self.options = self._extract_options()
self.non_negotiables = self._extract_non_negotiables()
self.criteria = self._extract_criteria()
self.facts = self._extract_facts()
self.assumptions = self._extract_assumptions()
self.fears = self._extract_fears()
self.recommended_option = self._recommend_choice()
self.review_date = self._review_date()
def _split_snippets(self) -> List[str]:
parts = re.split(r'[\n.;!?]+', self.raw)
return [part.strip(' -•') for part in parts if part.strip(' -•')]
def _extract_decision(self) -> str:
match = re.search(r'(should i .+?|choose between .+?|decide whether to .+?)(?:[.?!]|$)', self.lower)
if match:
decision = match.group(1)
elif self.snippets:
decision = self.snippets[0]
else:
decision = 'Choose the strongest path from the available options'
decision = decision.strip()
return decision[:1].upper() + decision[1:]
def _extract_deadline(self) -> str:
match = re.search(r'(by\s+[a-z0-9\- ]+|before\s+[a-z0-9\- ]+|next\s+week|this\s+month|this\s+quarter|tomorrow)', self.lower)
if match:
return match.group(1).strip()
return 'No hard deadline stated'
def _extract_options(self) -> List[str]:
patterns = [
r'between\s+(.+?)\s+and\s+(.+?)(?:[.?!]|$)',
r'should i\s+(.+?)\s+or\s+(.+?)(?:[.?!]|$)',
r'decide whether to\s+(.+?)\s+or\s+(.+?)(?:[.?!]|$)',
r'(.+?)\s+vs\.?\s+(.+?)(?:[.?!]|$)',
]
options: List[str] = []
for pattern in patterns:
match = re.search(pattern, self.lower)
if match:
options = [_clean_option(match.group(1)), _clean_option(match.group(2))]
break
if not options:
options = ['Stay with the current path', 'Make a smaller new move']
if not any('test' in option.lower() or 'trial' in option.lower() or 'pilot' in option.lower() for option in options):
options.append('Run a smaller reversible test')
return options[:3]
def _extract_non_negotiables(self) -> List[str]:
items: List[str] = []
if any(word in self.lower for word in ['family', 'kids', 'partner', 'home']):
items.append('Protect family stability and attention.')
if any(word in self.lower for word in ['money', 'salary', 'income', 'cash', 'runway', 'budget']):
items.append('Protect financial runway.')
if any(word in self.lower for word in ['health', 'sleep', 'burnout', 'stress']):
items.append('Avoid a choice that clearly harms health or recovery.')
if any(word in self.lower for word in ['values', 'ethic', 'integrity']):
items.append('Do not trade away core values for short-term gain.')
if not items:
items = ['Protect financial and emotional stability.', 'Keep the first move as reversible as possible.']
return items
def _extract_criteria(self) -> List[str]:
found = [label for label, keywords in CRITERIA_RULES.items() if any(keyword in self.lower for keyword in keywords)]
defaults = ['Cost and runway', 'Time and life load', 'Meaning and growth', 'Reversibility']
ordered: List[str] = []
for item in found + defaults:
if item not in ordered:
ordered.append(item)
return ordered[:4]
def _extract_facts(self) -> str:
for snippet in self.snippets:
if any(word in snippet.lower() for word in ['already', 'currently', 'have', 'is', 'are', 'deadline', 'offer']):
return snippet
return f'You are comparing {self.options[0]} and {self.options[1]}.'
def _extract_assumptions(self) -> str:
for snippet in self.snippets:
if any(word in snippet.lower() for word in ['maybe', 'probably', 'might', 'if', 'could']):
return snippet
return 'Some upside and downside estimates are still assumptions, not verified evidence.'
def _extract_fears(self) -> str:
for snippet in self.snippets:
if any(word in snippet.lower() for word in ['afraid', 'worry', 'fear', 'regret', 'fail']):
return snippet
return 'Fear of choosing the wrong branch permanently is louder than the available proof.'
def _reversibility(self, option: str) -> str:
lower = option.lower()
if any(word in lower for word in ['test', 'trial', 'pilot', 'part-time', 'part time', 'prototype']):
return 'Mostly reversible'
if any(word in lower for word in ['quit', 'resign', 'buy', 'move', 'full-time', 'full time']):
return 'Harder to reverse'
return 'Partly reversible'
def _key_unknown(self, option: str) -> str:
if 'Cost and runway' in self.criteria:
return 'The true cost and runway required.'
if 'Family impact' in self.criteria:
return 'The effect on family energy and logistics.'
if 'Meaning and growth' in self.criteria:
return 'Whether the option still feels meaningful after the novelty fades.'
return 'Which hidden tradeoff matters most in real life.'
def _branch_summary(self, option: str) -> Dict[str, str]:
reversible = self._reversibility(option)
best_case = f'{option} aligns with your top criteria and creates useful momentum.'
base_case = f'{option} works, but it comes with tradeoffs in {self.criteria[0].lower()} and {self.criteria[1].lower()}.'
worst_case = f'{option} looks attractive now, but the hidden costs are larger than expected.'
return {
'best': best_case,
'base': base_case,
'worst': worst_case,
'reversible': reversible,
'unknown': self._key_unknown(option),
}
def _recommend_choice(self) -> str:
test_option = next((option for option in self.options if 'test' in option.lower() or 'trial' in option.lower()), None)
if test_option and ('fear' in self.fears.lower() or self.deadline == 'No hard deadline stated'):
return test_option
stable_option = next((option for option in self.options if 'stay' in option.lower() or 'current' in option.lower()), None)
if stable_option and any('runway' in item.lower() or 'stability' in item.lower() for item in self.non_negotiables):
return stable_option
growth_option = next((option for option in self.options if any(word in option.lower() for word in ['start', 'build', 'move', 'create'])), None)
if growth_option and 'Meaning and growth' in self.criteria:
return growth_option
return self.options[0]
def _review_date(self) -> str:
if self.deadline != 'No hard deadline stated':
return self.deadline
return 'After one real-world test or within 7 days'
def render(self) -> str:
lines: List[str] = []
lines.append('# Decision Forest')
lines.append('')
lines.append('## Decision Trunk')
lines.append(f'- Decision: {self.decision}')
lines.append(f'- Deadline: {self.deadline}')
lines.append(f"- Non-negotiables: {'; '.join(self.non_negotiables)}")
lines.append(f'- Facts: {self.facts}')
lines.append(f'- Assumptions: {self.assumptions}')
lines.append(f'- Fears: {self.fears}')
lines.append('')
lines.append('## Criteria')
for criterion in self.criteria:
lines.append(f'- {criterion}')
lines.append('')
lines.append('## Branches')
labels = ['A', 'B', 'C']
for label, option in zip(labels, self.options):
summary = self._branch_summary(option)
lines.append(f'### Option {label}: {option}')
lines.append(f"- Best case: {summary['best']}")
lines.append(f"- Base case: {summary['base']}")
lines.append(f"- Worst case: {summary['worst']}")
lines.append(f"- Reversible or not: {summary['reversible']}")
lines.append(f"- Key unknown: {summary['unknown']}")
lines.append('')
lines.append('## Choice Logic')
lines.append(f'- Most aligned option: {self.recommended_option}')
lines.append('- Why: It protects learning while keeping the biggest constraints visible.')
lines.append('- What would change the choice: Clearer evidence on the main unknown or a non-negotiable becoming more urgent.')
lines.append(f'- Review date: {self.review_date}')
return '\n'.join(lines)
def handle(inputs):
_load_skill_meta('decision-forest')
planner = DecisionForestPlanner(inputs)
return planner.render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import DecisionForestPlanner, handle
def test_option_extraction():
planner = DecisionForestPlanner('Should I stay in my job or start a small consulting business by June?')
assert 'Stay in my job' in planner.options[0]
assert any('test' in option.lower() for option in planner.options)
def test_recommended_option_exists():
planner = DecisionForestPlanner('I worry about money, so I need help deciding whether to stay or start something new.')
assert planner.recommended_option
def test_output_sections():
output = handle('Should I move cities or stay where I am? I am afraid of regretting it later.')
assert output.startswith('# Decision Forest')
assert '- Facts:' in output
assert '- Assumptions:' in output
assert '- Fears:' in output
assert '- Review date:' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Guide a weekly reflection across Work, Health, Relationships, and Growth, then turn the review into one clear pattern plus Stop, Start, Continue, and one exp...
---
name: weekly-reflectron
description: Guide a weekly reflection across Work, Health, Relationships, and Growth, then turn the review into one clear pattern plus Stop, Start, Continue, and one experiment for next week. Use when the user wants a structured weekly review that is concrete, balanced, and actionable instead of vague or self-critical.
---
# Weekly Reflectron
## Overview
Use this skill to review a week through four life dimensions instead of collapsing everything into work stress or generic self-judgment. It helps the user gather evidence first, spot one cross-cutting pattern, and convert insight into a small adjustment for the next week.
This skill is descriptive only. It does not sync calendars, task apps, fitness devices, or journals.
## Trigger
Use this skill when the user wants to:
- do a weekly review that covers more than work
- capture wins and friction with concrete evidence
- reduce vague, harsh, or repetitive reflection
- produce Stop, Start, Continue actions for the coming week
- end a review with one experiment instead of a total reinvention
### Example prompts
- "Help me do a weekly reflection without turning it into self-criticism"
- "Give me a five minute weekly review for work, health, relationships, and growth"
- "Turn this messy week into a useful reflection"
- "I want a weekly reset with one clear experiment for next week"
## Workflow
1. Review the week through Work, Health, Relationships, and Growth.
2. Gather evidence first, including wins, slips, habits, and emotional highs or lows.
3. Label each dimension with a simple status.
4. Notice one pattern that touched multiple parts of the week.
5. Convert the pattern into one Stop, one Start, and one Continue action.
6. End with one experiment for next week.
## Inputs
The user can provide any mix of:
- freeform notes about the week
- work progress or unfinished loops
- sleep, food, movement, or energy notes
- family, partner, or friendship moments
- learning, reading, journaling, or growth notes
- overall mood or stress level
- explicit weekly wins and friction points
## Outputs
Return a markdown reflection with:
- snapshot of the week
- four-dimension review
- one pattern noticed
- Stop, Start, Continue, and one experiment for next week
## Safety
- Use concrete evidence when available, not abstract self-judgment.
- During crisis or overload weeks, optimize for stabilization before ambition.
- Balance wins and friction. Do not turn the review into a problem-only narrative.
- Do not claim psychological certainty from limited notes.
## Acceptance Criteria
- Return markdown text.
- Cover Work, Health, Relationships, and Growth.
- Include at least one actionable next-week adjustment.
- Keep the reflection concrete, balanced, and usable.
FILE:handler.py
#!/usr/bin/env python3
import json
import os
import re
import sys
from typing import Any, Dict, List
def _load_skill_meta(slug):
path = os.path.join(os.path.dirname(__file__), 'SKILL.md')
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
meta = re.search(r'^---$(.*?)^---$', content, re.DOTALL | re.MULTILINE)
return meta.group(1).strip() if meta else ''
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ''
if isinstance(inputs, str):
return inputs.strip()
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
DIMENSION_KEYWORDS = {
'Work': ['work', 'project', 'client', 'meeting', 'deadline', 'task', 'email', 'draft', 'study', 'school'],
'Health': ['sleep', 'walk', 'workout', 'exercise', 'meal', 'food', 'water', 'rest', 'run', 'tired', 'energy'],
'Relationships': ['family', 'friend', 'partner', 'child', 'kids', 'parent', 'call', 'conversation', 'date'],
'Growth': ['read', 'book', 'learn', 'learning', 'course', 'journal', 'reflect', 'practice', 'skill'],
}
POSITIVE_WORDS = ['good', 'great', 'win', 'finished', 'progress', 'proud', 'steady', 'strong', 'connected', 'consistent', 'better']
NEGATIVE_WORDS = ['hard', 'late', 'stuck', 'avoid', 'avoided', 'missed', 'tired', 'poor', 'friction', 'stress', 'overwhelmed']
DEFAULT_WINS = {
'Work': 'Name one concrete thing that moved, even if the week felt messy.',
'Health': 'Name one behavior that supported your body, even briefly.',
'Relationships': 'Name one moment of contact, care, or patience.',
'Growth': 'Name one thing you learned, noticed, or practiced.',
}
DEFAULT_FRICTIONS = {
'Work': 'Notice one unfinished loop or recurring distraction.',
'Health': 'Notice one pattern around sleep, food, movement, or energy that made the week harder.',
'Relationships': 'Notice one relationship that got less attention than it needed.',
'Growth': 'Notice where learning or reflection got crowded out.',
}
DEFAULT_EVIDENCE = {
'Work': 'No work note was provided, so pull one concrete example from memory.',
'Health': 'No health note was provided, so pull one real body or energy example from memory.',
'Relationships': 'No relationship note was provided, so pull one interaction from memory.',
'Growth': 'No growth note was provided, so pull one reading, learning, or reflection example from memory.',
}
class WeeklyReflector:
def __init__(self, inputs: Any):
self.raw = _normalize_inputs(inputs)
self.lower = self.raw.lower()
self.snippets = self._split_snippets()
self.dimensions = self._build_dimensions()
self.pattern = self._detect_pattern()
self.theme = self._detect_theme()
self.overall_feeling = self._overall_feeling()
self.adjustments = self._build_adjustments()
def _split_snippets(self) -> List[str]:
parts = re.split(r'[\n.;!?]+', self.raw)
return [part.strip(' -•') for part in parts if part.strip(' -•')]
def _matched_snippets(self, dimension: str) -> List[str]:
keywords = DIMENSION_KEYWORDS[dimension]
matched = [snippet for snippet in self.snippets if any(keyword in snippet.lower() for keyword in keywords)]
return matched
def _positive_snippets(self, snippets: List[str]) -> List[str]:
return [snippet for snippet in snippets if any(word in snippet.lower() for word in POSITIVE_WORDS)]
def _negative_snippets(self, snippets: List[str]) -> List[str]:
return [snippet for snippet in snippets if any(word in snippet.lower() for word in NEGATIVE_WORDS)]
def _build_dimensions(self) -> Dict[str, Dict[str, str]]:
result: Dict[str, Dict[str, str]] = {}
for dimension in DIMENSION_KEYWORDS:
matched = self._matched_snippets(dimension)
positives = self._positive_snippets(matched)
negatives = self._negative_snippets(matched)
if positives and not negatives:
status = 'strong'
elif negatives and not positives:
status = 'needs attention'
elif matched:
status = 'mixed'
else:
status = 'mixed'
wins = '; '.join(positives[:2]) if positives else DEFAULT_WINS[dimension]
friction = '; '.join(negatives[:2]) if negatives else DEFAULT_FRICTIONS[dimension]
evidence = '; '.join(matched[:3]) if matched else DEFAULT_EVIDENCE[dimension]
result[dimension] = {
'status': status,
'wins': wins,
'friction': friction,
'evidence': evidence,
}
return result
def _detect_pattern(self) -> str:
if any(word in self.lower for word in ['sleep', 'slept poorly', 'late night', 'tired', 'exhausted']):
return 'Sleep and recovery shaped multiple areas of the week more than willpower alone.'
if any(word in self.lower for word in ['avoid', 'avoided', 'procrast', 'stuck', 'delay']):
return 'Avoidance and delayed starts created drag across more than one area.'
strong_count = sum(1 for info in self.dimensions.values() if info['status'] == 'strong')
if strong_count >= 3 or any(word in self.lower for word in ['consistent', 'routine', 'showed up', 'kept going']):
return 'Small consistency created momentum across the week.'
if self.dimensions['Relationships']['status'] == 'needs attention':
return 'Relationships needed more intentional space instead of being squeezed between tasks.'
return 'The week looked mixed, so one stabilizing adjustment will likely help more than a full reset.'
def _detect_theme(self) -> str:
if self.pattern.startswith('Sleep'):
return 'Recovery'
if self.pattern.startswith('Small consistency'):
return 'Momentum'
if self.pattern.startswith('Avoidance'):
return 'Re-entry'
return 'Rebalance'
def _overall_feeling(self) -> str:
positives = sum(1 for snippet in self.snippets if any(word in snippet.lower() for word in POSITIVE_WORDS))
negatives = sum(1 for snippet in self.snippets if any(word in snippet.lower() for word in NEGATIVE_WORDS))
if positives > negatives:
return 'Mostly positive, with usable momentum.'
if negatives > positives:
return 'Tired or stretched, but still informative.'
return 'Mixed, honest, and still useful.'
def _build_adjustments(self) -> Dict[str, str]:
if self.pattern.startswith('Sleep'):
return {
'stop': 'Stop borrowing from sleep to rescue the next day.',
'start': 'Start one earlier shutdown cue on the two most demanding nights.',
'continue': 'Continue the smallest health behavior that already helped, even if it was brief.',
'experiment': 'Try a two-night wind-down experiment and note whether work focus and patience improve.',
}
if self.pattern.startswith('Avoidance'):
return {
'stop': 'Stop carrying vague open loops in your head.',
'start': 'Start the day with one 15 minute entry task on the hardest item.',
'continue': 'Continue anything that made starting easier, even if the block was short.',
'experiment': 'Test a visible first-task card for three days and see if resistance drops.',
}
if self.pattern.startswith('Small consistency'):
return {
'stop': 'Stop overloading next week with too many new goals.',
'start': 'Start protecting the routines that already produced momentum.',
'continue': 'Continue the habits and environments that made follow-through easier.',
'experiment': 'Repeat your strongest block from this week on purpose once more next week.',
}
return {
'stop': 'Stop keeping the weekly review only in your head.',
'start': 'Start a 10 minute weekly evidence capture before judging the week.',
'continue': 'Continue one supportive behavior that already feels realistic.',
'experiment': 'Run one small change in the area with the most friction and review it next week.',
}
def render(self) -> str:
lines: List[str] = []
lines.append('# Weekly Reflection')
lines.append('')
lines.append('## Snapshot')
lines.append(f'- Week theme: {self.theme}')
lines.append(f'- Overall feeling: {self.overall_feeling}')
lines.append('')
lines.append('## Four-Dimension Review')
for dimension in ['Work', 'Health', 'Relationships', 'Growth']:
info = self.dimensions[dimension]
lines.append(f'### {dimension}')
lines.append(f"- Status: {info['status']}")
lines.append(f"- Wins: {info['wins']}")
lines.append(f"- Friction: {info['friction']}")
lines.append(f"- Evidence: {info['evidence']}")
lines.append('')
lines.append('## Pattern Noticed')
lines.append(f'- {self.pattern}')
lines.append('')
lines.append('## Next Week Adjustments')
lines.append(f"- Stop: {self.adjustments['stop']}")
lines.append(f"- Start: {self.adjustments['start']}")
lines.append(f"- Continue: {self.adjustments['continue']}")
lines.append(f"- One experiment: {self.adjustments['experiment']}")
return '\n'.join(lines)
def handle(inputs):
_load_skill_meta('weekly-reflectron')
reflector = WeeklyReflector(inputs)
return reflector.render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import WeeklyReflector, handle
def test_sleep_pattern_detection():
reflector = WeeklyReflector('Poor sleep made work harder, but I still finished a draft and went for two walks.')
assert reflector.pattern.startswith('Sleep and recovery')
def test_dimension_coverage():
reflector = WeeklyReflector('Great client progress. Family dinner felt connected. Read two chapters of a book.')
assert set(reflector.dimensions.keys()) == {'Work', 'Health', 'Relationships', 'Growth'}
def test_output_sections():
output = handle('Finished a client draft, slept poorly, and had a good talk with my partner.')
assert output.startswith('# Weekly Reflection')
assert '### Work' in output
assert '### Health' in output
assert '### Relationships' in output
assert '### Growth' in output
assert '- One experiment:' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Build a realistic morning routine around the user's real wake time, time budget, reliable anchor, night-before prep, and fallback plan for chaotic or low-sle...
---
name: morning-ritual-planner
description: Build a realistic morning routine around the user's real wake time, time budget, reliable anchor, night-before prep, and fallback plan for chaotic or low-sleep mornings. Use when the user wants a calmer, clearer, or more energized start without copying unrealistic influencer routines or relying on app automation.
---
# Morning Ritual Planner
## Overview
Use this skill to turn a vague wish for a better morning into a practical ritual that matches real life. It keeps the routine honest by starting from the user's actual wake time, existing obligations, and one anchor they already perform reliably.
This skill is descriptive only. It does not set alarms, start timers, or connect to calendars, wearables, or reminder apps.
## Trigger
Use this skill when the user wants to:
- design a morning ritual that fits a 10, 20, 30, or 45 minute window
- stop overbuilding a routine they cannot sustain
- adapt a morning plan to family, commute, caregiving, or work constraints
- create a shorter fallback version for late or low-sleep mornings
- add night-before prep so the routine feels easier to start
### Example prompts
- "Help me build a realistic 20 minute morning routine before the school run"
- "I want a calmer morning, but I only have 10 minutes"
- "Create a fallback morning ritual for days when I wake up tired"
- "Design a focused morning routine around coffee and journaling"
## Workflow
1. Capture the user's real wake time, desired morning feeling, and hard constraints.
2. Identify a reliable anchor, such as getting out of bed, brushing teeth, or making coffee.
3. Build the routine in layers: must-do, nice-to-have, and bonus.
4. Fit the sequence to an honest time budget.
5. Add one or more night-before prep steps that reduce friction.
6. Create a fallback version for chaotic or low-energy mornings.
7. End with a short weekly review prompt so the routine can evolve.
## Inputs
The user can provide any mix of:
- wake time or target wake window
- available morning time
- desired feeling, such as calm, focused, grounded, or energized
- existing anchor habit
- family, commute, work, school, or caregiving constraints
- preferred activities, such as stretching, tea, journaling, reading, or movement
- sleep quality or low-energy context
## Outputs
Return a markdown plan with:
- goal summary
- wake-up anchor and honest time budget
- ritual sequence with durations and why each step stays
- night-before prep list
- fallback version for hard mornings
- weekly review questions
## Safety
- Keep the plan realistic for the user's current life, not an aspirational fantasy schedule.
- Protect sleep. If the user sounds chronically sleep deprived, shorten the ritual before adding more activities.
- Prefer interruptible routines for caregivers or chaotic households.
- Do not claim the routine will guarantee productivity, mood, or discipline.
## Acceptance Criteria
- Return markdown text.
- Include one reliable anchor, one night-before prep step, and one fallback version.
- Keep the total duration realistic for the stated time budget.
- Make it obvious what a successful normal morning looks like.
FILE:handler.py
#!/usr/bin/env python3
import json
import os
import re
import sys
from typing import Any, List, Tuple
def _load_skill_meta(slug):
path = os.path.join(os.path.dirname(__file__), 'SKILL.md')
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
meta = re.search(r'^---$(.*?)^---$', content, re.DOTALL | re.MULTILINE)
return meta.group(1).strip() if meta else ''
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ''
if isinstance(inputs, str):
return inputs.strip()
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _dedupe(items: List[str]) -> List[str]:
seen = set()
ordered: List[str] = []
for item in items:
if item not in seen:
seen.add(item)
ordered.append(item)
return ordered
FEELING_RULES = {
'calm and grounded': ['calm', 'peace', 'grounded', 'gentle', 'steady', 'soft'],
'focused and clear': ['focus', 'focused', 'clarity', 'clear', 'productive', 'sharp'],
'energized and upbeat': ['energ', 'active', 'wake up', 'wake-up', 'strong', 'workout', 'upbeat'],
}
ANCHOR_RULES = [
('after school drop-off', ['drop-off', 'school run', 'school drop off']),
('brush teeth', ['brush teeth', 'toothbrush', 'brushing']),
('make coffee', ['coffee', 'espresso']),
('make tea', ['tea', 'kettle']),
('drink water', ['drink water', 'water bottle', 'glass of water']),
('get out of bed', ['get out of bed', 'wake up', 'alarm']),
]
CONSTRAINT_RULES = [
('Caregiving or school logistics may interrupt the flow.', ['kids', 'kid', 'child', 'children', 'caregiver', 'school']),
('The morning needs a fast handoff into commute or work.', ['commute', 'office', 'train', 'out the door', 'meeting']),
('Energy may be limited after poor sleep.', ['poor sleep', 'tired', 'exhausted', 'sleep deprived', 'slept poorly']),
('A rotating or variable schedule may need multiple versions.', ['rotating', 'shift', 'variable schedule', 'different every day']),
]
RITUAL_TEMPLATES = {
'calm and grounded': [
('Drink water and open the curtains', 'It wakes the body gently and gives the morning a visible starting point.'),
('Take two slow minutes for breathing, prayer, or silence', 'It lowers mental noise before the day begins to pull on you.'),
('Do light stretching or slow mobility', 'It turns stiffness into steadiness without needing much willpower.'),
('Write the one thing that matters most today', 'It keeps the day intentional instead of reactive.'),
('Have tea and read one quiet page or journal a few lines', 'It adds a human rhythm instead of a rushed mechanical start.'),
('Take a longer walk or reflective journaling block', 'It deepens the sense of grounded momentum when time allows.'),
],
'focused and clear': [
('Drink water and stand in bright light', 'It helps the brain switch into awake mode quickly.'),
('Review today\'s top priority', 'It protects focus before notifications take over.'),
('Do quick mobility or a brisk walk', 'It raises alertness without turning the morning into a full workout.'),
('Write three lines in a notebook', 'It clears mental tabs and reduces background clutter.'),
('Preview the first work or study block', 'It makes the next meaningful action obvious.'),
('Do a longer planning or learning block', 'It converts spare time into deliberate progress.'),
],
'energized and upbeat': [
('Drink water and turn on bright light or upbeat music', 'It gives the body a quick activation signal.'),
('Take three deep breaths while standing tall', 'It creates a physical shift without adding complexity.'),
('Do mobility, a brisk walk, or light cardio', 'It builds energy fast and prevents a sluggish start.'),
('Use a quick shower or face splash reset', 'It sharpens the transition from sleepy to ready.'),
('Add a short workout or protein-forward breakfast step', 'It makes the energy boost more durable.'),
('Do a fuller movement block before the day opens up', 'It uses the longer window for real momentum instead of scatter.'),
],
}
DURATION_MAP = {
10: [2, 3, 5],
20: [2, 3, 5, 10],
30: [2, 3, 5, 8, 12],
45: [2, 3, 10, 10, 8, 12],
}
LAYER_LABELS = ['Must-do', 'Must-do', 'Nice-to-have', 'Nice-to-have', 'Bonus', 'Bonus']
class MorningRitualPlanner:
def __init__(self, inputs: Any):
self.raw = _normalize_inputs(inputs)
self.lower = self.raw.lower()
self.time_budget = self._extract_time_budget()
self.wake_time = self._extract_wake_time()
self.anchor = self._detect_anchor()
self.desired_feeling = self._detect_feeling()
self.constraints = self._detect_constraints()
self.sequence = self._build_sequence()
self.prep_steps = self._night_before_prep()
self.fallback = self._fallback_version()
def _extract_time_budget(self) -> int:
match = re.search(r'(\d{1,2})\s*(?:minutes|minute|mins|min)\b', self.lower)
if match:
return int(match.group(1))
if any(word in self.lower for word in ['quick', 'tiny window', 'short', 'tight']):
return 10
if any(word in self.lower for word in ['half hour', '30 minute', '30-minute']):
return 30
if any(word in self.lower for word in ['long', 'full routine', '45 minute', '45-minute']):
return 45
return 20
def _extract_wake_time(self) -> str:
match = re.search(r'(?:wake(?: up)?(?: at| around| by)?|alarm(?: for)?|up by)\s*(\d{1,2}(?::\d{2})?\s*(?:am|pm)?)', self.lower)
if match:
return match.group(1).strip()
return 'Not specified'
def _detect_anchor(self) -> str:
for label, keywords in ANCHOR_RULES:
if any(keyword in self.lower for keyword in keywords):
return label
return 'get out of bed'
def _detect_feeling(self) -> str:
for label, keywords in FEELING_RULES.items():
if any(keyword in self.lower for keyword in keywords):
return label
return 'calm and grounded'
def _detect_constraints(self) -> List[str]:
found = [message for message, keywords in CONSTRAINT_RULES if any(keyword in self.lower for keyword in keywords)]
if self.time_budget <= 10:
found.append('The routine has to fit a genuinely small morning window.')
if not found:
found.append('No major morning constraint was specified, so the routine stays simple and flexible.')
return _dedupe(found)
def _bucket(self) -> int:
if self.time_budget <= 10:
return 10
if self.time_budget <= 20:
return 20
if self.time_budget <= 30:
return 30
return 45
def _build_sequence(self) -> List[Tuple[str, str, int, str]]:
template = RITUAL_TEMPLATES[self.desired_feeling]
durations = DURATION_MAP[self._bucket()]
sequence: List[Tuple[str, str, int, str]] = []
for index, duration in enumerate(durations):
activity, why = template[index]
layer = LAYER_LABELS[index]
sequence.append((layer, activity, duration, why))
return sequence
def _night_before_prep(self) -> List[str]:
prep = [
'Lay out the first object the ritual needs, such as clothes, notebook, water bottle, or mat.',
'Decide what the first screen boundary is, so the morning does not begin inside notifications.',
]
if self.anchor in ('make coffee', 'make tea'):
prep.append('Set out the mug, kettle, and coffee or tea setup before bed.')
elif self.anchor == 'brush teeth':
prep.append('Place a filled water bottle or notebook near the sink so the next step is automatic.')
elif self.anchor == 'after school drop-off':
prep.append('Pack bags, clothes, and anything needed for the school handoff the night before.')
else:
prep.append('Choose the first two ritual steps in advance so there is no morning negotiation.')
return prep
def _fallback_version(self) -> str:
if any('poor sleep' in item.lower() or 'energy may be limited' in item.lower() for item in self.constraints):
return 'keep only water, three slow breaths, and the first must-do step, then skip every bonus item without guilt.'
if self.time_budget <= 10:
return 'keep only the anchor, one calming or energizing reset, and the first priority cue.'
return 'compress the ritual into a 3 minute version: anchor, water, one short reset, and a single next-step cue.'
def render(self) -> str:
lines: List[str] = []
lines.append('# Morning Ritual Plan')
lines.append('')
lines.append('## Goal')
lines.append(f'- Desired morning feeling: {self.desired_feeling}')
lines.append(f'- Available time: {self.time_budget} minutes')
lines.append(f'- Wake-up anchor: {self.anchor}')
lines.append(f'- Real wake time: {self.wake_time}')
lines.append(f"- Real-world constraints: {'; '.join(self.constraints)}")
lines.append('')
lines.append('## Ritual Sequence')
for index, (layer, activity, duration, why) in enumerate(self.sequence, 1):
lines.append(f'{index}. [{layer}] {activity} - {duration} min - {why}')
lines.append('')
lines.append('## Night-Before Prep')
for item in self.prep_steps:
lines.append(f'- {item}')
lines.append('')
lines.append('## Fallback Version')
lines.append(f'- If I woke late or slept poorly, I will: {self.fallback}')
lines.append('')
lines.append('## Weekly Review')
lines.append('- What felt easy?')
lines.append('- What created friction?')
lines.append('- What should be shortened, moved, or removed?')
return '\n'.join(lines)
def handle(inputs):
_load_skill_meta('morning-ritual-planner')
planner = MorningRitualPlanner(inputs)
return planner.render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import MorningRitualPlanner, handle
def test_time_budget_detection():
planner = MorningRitualPlanner('I only have 10 minutes before the school run.')
assert planner.time_budget == 10
def test_anchor_detection():
planner = MorningRitualPlanner('After I make coffee I want a calmer start to the day.')
assert planner.anchor == 'make coffee'
def test_output_sections():
output = handle('I want a focused 20 minute routine after I brush my teeth.')
assert output.startswith('# Morning Ritual Plan')
assert '## Fallback Version' in output
assert '## Weekly Review' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Build a settlement reconciliation brief for marketplace, PSP, and bank payout workflows. Use when a team needs to separate true cash discrepancies from timin...
---
name: settlement-reconciliation-guard
description: Build a settlement reconciliation brief for marketplace, PSP, and bank payout workflows. Use when a team needs to separate true cash discrepancies from timing effects, fee issues, refund timing, reserve leakage, or order-to-payout mapping gaps.
---
# Settlement Reconciliation Guard
## Overview
Settlement Reconciliation Guard turns a short finance or operations prompt into a structured reconciliation-control brief.
It is useful when you need a fast view of likely discrepancy themes, evidence requirements, matching logic, and close controls for payout reviews.
## Use this skill when
- daily or weekly settlement checks are surfacing unexplained differences
- month-end close needs a tighter review of payout, fee, refund, reserve, or tax treatment
- a team is investigating a missing payout, short payment, or unreconciled batch
- a new marketplace, PSP, or payout channel is being onboarded and needs control design
- finance needs a concise investigation brief before escalating to a provider or posting journals
## What the skill does
The handler reads the prompt, infers likely review context, and produces a structured brief with:
1. **Review mode** such as daily payout check, weekly exception sweep, month-end close support, discrepancy investigation, or new channel setup review
2. **Channels in scope** such as marketplace settlement, PSP/card acquiring, bank remittance, or ERP ledger workflows
3. **Evidence referenced** such as order ledgers, payout reports, bank statements, fee schedules, refund logs, reserve logs, or tax files
4. **Primary discrepancy themes** such as missing payout, fee mismatch, refund timing mismatch, reserve leakage, order-to-payout mapping gaps, or tax variance
5. **Investigation priorities and close controls** that help separate timing differences from true financial loss
## Recommended input patterns
Use plain language. Helpful details include:
- review cadence or close context
- channel or provider involved
- evidence sources available
- known discrepancy symptoms
- whether the issue is a routine review, escalation, or new-channel setup
Example prompts:
- `Need month-end close support for payout and refund timing review.`
- `Investigate missing payout, fee mismatch, and unreconciled batch differences.`
- `Compare Amazon payout report with bank statement and fee schedule before posting to the ERP general ledger.`
- `Create a weekly exception sweep for refund, reserve, and fee variance issues.`
## Output shape
The skill returns a markdown brief with sections such as:
- Reconciliation Summary
- Matching Logic Table
- Discrepancy Queue
- Investigation Priorities
- Close Controls
- Assumptions and Limits
## Boundaries
- This skill is heuristic. It does not query live settlement portals, PSP dashboards, bank feeds, or ERP systems.
- It helps structure reconciliation thinking and evidence review, not replace accounting judgment or provider-side investigation.
- Final journal treatment, dispute escalation, reserve handling, and approval decisions should remain human-approved.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
REVIEW_MODE_RULES = {
"Daily payout check": ["daily", "yesterday", "daily settlement", "payout run"],
"Weekly exception sweep": ["weekly", "exception sweep", "weekly review"],
"Month-end close support": ["month-end", "month end", "close", "quarter close"],
"Discrepancy investigation": ["discrepancy", "mismatch", "missing payout", "short paid", "unreconciled", "investigate"],
"New channel setup review": ["new channel", "new psp", "new marketplace", "launch", "onboarding"],
}
CHANNEL_RULES = {
"Marketplace settlement": ["amazon", "tiktok", "jd", "taobao", "marketplace"],
"PSP / card acquiring": ["psp", "stripe", "adyen", "paypal", "card", "acquirer"],
"Bank payout / remittance": ["bank", "statement", "remittance", "payout file"],
"ERP / finance ledger": ["erp", "general ledger", "gl", "finance ledger", "journal"],
}
EVIDENCE_RULES = {
"Order ledger": ["order ledger", "orders", "merchant ledger", "order report"],
"Payout report": ["payout", "settlement report", "remittance advice", "provider report"],
"Bank statement": ["bank statement", "statement", "bank file"],
"Fee schedule": ["fee schedule", "mdr", "commission", "fee table", "pricing schedule"],
"Refund log": ["refund", "refund log", "refund report"],
"Chargeback or reserve log": ["chargeback", "reserve", "holdback", "dispute"],
"Tax or invoice file": ["tax", "vat", "invoice", "fapiao"],
}
DISCREPANCY_RULES = {
"Missing payout or delayed remittance": {
"keywords": ["missing payout", "delayed payout", "remittance", "not received", "bank delay"],
"why": "Cash expected from the channel has not landed when finance expects it to clear.",
"check": "Compare payout schedule, processor release rules, and the bank statement timing before escalating a shortage claim.",
"action": "Tag the batch, prove the expected amount, and separate timing delay from actual loss.",
},
"Fee mismatch": {
"keywords": ["fee mismatch", "commission", "mdr", "higher fee", "rate error", "fee"],
"why": "Incorrect fee application distorts margin and makes every downstream settlement look off.",
"check": "Reconcile contracted rates, fee tiers, promotions, and tax treatment against the payout detail.",
"action": "Build a fee-variance file and isolate systemic pricing-rule errors from one-off manual adjustments.",
},
"Refund timing mismatch": {
"keywords": ["refund", "refund timing", "returned", "reversal", "refund delay"],
"why": "Refunds often hit a different cycle than sales and can make a clean period look broken.",
"check": "Match refund authorization date, customer completion date, and settlement-cycle impact separately.",
"action": "Keep a cross-period refund schedule so finance does not double-count or chase normal timing differences.",
},
"Chargeback or reserve leakage": {
"keywords": ["chargeback", "reserve", "holdback", "dispute", "rolling reserve"],
"why": "Dispute and reserve mechanisms quietly withhold cash outside the core sales-versus-payout view.",
"check": "Review reserve policy, release timing, and dispute workflow before calling it a settlement shortfall.",
"action": "Split reserves and chargebacks into dedicated reconciliation buckets with named owners.",
},
"Order-to-payout mapping gap": {
"keywords": ["mapping", "unmatched", "order missing", "sku missing", "join key", "reconcile"],
"why": "When identifiers do not join cleanly, finance teams can neither explain the gap nor prove the true balance.",
"check": "Validate order ID, payout batch, currency, and merchant reference keys across every source file.",
"action": "Repair the join logic first, then investigate residual true discrepancies.",
},
"Tax or invoice variance": {
"keywords": ["tax", "vat", "invoice", "fapiao", "withholding"],
"why": "Tax treatment can legitimately change settlement math, but it is often misclassified as fee error or short payment.",
"check": "Confirm invoice rules, withholding logic, and whether tax is grossed up or netted in the report.",
"action": "Move tax review into the reconciliation checklist instead of handling it as an afterthought.",
},
}
class SettlementReconciliationGuard:
def __init__(self, inputs: Any):
self.meta = _load_skill_meta()
self.raw = inputs
self.text = _normalize_inputs(inputs)
self.lower = self.text.lower()
self.review_mode = _detect_one(self.lower, REVIEW_MODE_RULES, "Daily payout check")
self.channels = _detect_many(self.lower, CHANNEL_RULES, ["Marketplace settlement", "Bank payout / remittance"], limit=4)
self.evidence = _detect_many(self.lower, EVIDENCE_RULES, ["Order ledger", "Payout report"], limit=5)
self.discrepancies = self._detect_discrepancies()
def _detect_discrepancies(self) -> List[str]:
matched = [name for name, data in DISCREPANCY_RULES.items() if any(keyword in self.lower for keyword in data["keywords"])]
defaults = {
"Daily payout check": ["Missing payout or delayed remittance", "Order-to-payout mapping gap", "Fee mismatch"],
"Weekly exception sweep": ["Order-to-payout mapping gap", "Fee mismatch", "Refund timing mismatch"],
"Month-end close support": ["Refund timing mismatch", "Fee mismatch", "Tax or invoice variance"],
"Discrepancy investigation": ["Missing payout or delayed remittance", "Fee mismatch", "Order-to-payout mapping gap"],
"New channel setup review": ["Order-to-payout mapping gap", "Fee mismatch", "Tax or invoice variance"],
}[self.review_mode]
ordered: List[str] = []
for item in matched + defaults:
if item not in ordered:
ordered.append(item)
return ordered[:4]
def _matching_logic_rows(self) -> List[List[str]]:
return [
["Sales capture", "Order ledger to settlement detail", "Missing merchant references, currency drift, or split-order handling", "Make the join keys explicit before arguing about the money."],
["Fee application", "Contracted fee schedule to payout deductions", "Tier logic, promo rates, reserve deductions, or tax netting", "Prove expected fees separately from net cash movement."],
["Refund and reversal timing", "Refund log to payout cycle", "Cross-period timing and partial reversals", "Track timing buckets so expected reversals do not appear as unexplained loss."],
["Cash receipt", "Payout report to bank statement", "Bank delay, remittance timing, or reference mismatch", "Escalate to the provider only after the expected bank landing date is evidenced."],
]
def _priorities(self) -> List[str]:
items = [
"Confirm the batch scope first, which channel, currency, date range, and settlement cycle are actually under review.",
"Reconcile identifiers before reviewing commercial logic, because broken joins create fake fee and payout problems.",
"Escalate only the residual exceptions that remain after timing, reserve, and refund mechanics are separated.",
]
if "Bank statement" not in self.evidence:
items.append("Request bank evidence before declaring a cash shortfall, because many payout disputes are really timing disputes.")
return items[:4]
def _close_controls(self) -> List[str]:
return [
"Keep one owner for source-file completeness and one owner for reconciliation logic when finance and operations both touch the workflow.",
"Version the fee schedule and payout logic used for the reconciliation so month-end restatements are explainable later.",
"Store exception decisions with evidence links so the next review cycle starts from confirmed facts instead of memory.",
"Treat manual write-offs, reserve releases, and tax adjustments as separate approvals rather than hiding them in one balancing line.",
]
def _assumptions(self) -> List[str]:
notes = [
"This brief is heuristic and uses only the user-supplied context plus the local skill description.",
"No live settlement portal, bank feed, ERP, or PSP dashboard was queried.",
"Final accounting treatment, dispute escalation, and journal posting remain human-approved.",
]
missing = [item for item in ["Fee schedule", "Bank statement", "Refund log"] if item not in self.evidence]
if missing:
notes.append(f"Lightly referenced evidence sources: {_join(missing)}.")
return notes
def render(self) -> str:
lines: List[str] = []
lines.append("# Settlement Reconciliation Guard")
lines.append("")
lines.append(f"**Skill description:** {self.meta['description']}")
lines.append(f"**Review mode:** {self.review_mode}")
lines.append(f"**Channels in scope:** {_join(self.channels)}")
lines.append(f"**Primary discrepancy themes:** {_join(self.discrepancies)}")
lines.append(f"**Evidence referenced:** {_join(self.evidence)}")
lines.append(f"**Input snapshot:** {self.text or 'No structured settlement context was provided, so this brief uses default reconciliation-control assumptions.'}")
lines.append("")
lines.append("## Reconciliation Summary")
lines.append("- Separate timing differences from true settlement errors before escalating a financial discrepancy.")
lines.append("- Reconciliation works best when order, fee, refund, reserve, and cash-receipt logic are reviewed as distinct layers.")
lines.append(f"- Because the main review mode is **{self.review_mode.lower()}**, the output emphasizes exception containment and evidence quality over premature root-cause certainty.")
lines.append("")
lines.append("## Matching Logic Table")
lines.append("| Match stage | What to align | Common breakpoints | Control note |")
lines.append("|---|---|---|---|")
for row in self._matching_logic_rows():
lines.append(f"| {row[0]} | {row[1]} | {row[2]} | {row[3]} |")
lines.append("")
lines.append("## Discrepancy Queue")
lines.append("| Discrepancy theme | Why it matters | First check | Default response |")
lines.append("|---|---|---|---|")
for name in self.discrepancies:
info = DISCREPANCY_RULES[name]
lines.append(f"| {name} | {info['why']} | {info['check']} | {info['action']} |")
lines.append("")
lines.append("## Investigation Priorities")
for idx, item in enumerate(self._priorities(), 1):
lines.append(f"{idx}. {item}")
lines.append("")
lines.append("## Close Controls")
for item in self._close_controls():
lines.append(f"- {item}")
lines.append("")
lines.append("## Assumptions and Limits")
for note in self._assumptions():
lines.append(f"- {note}")
return "\n".join(lines)
def handle(inputs: Any) -> str:
return SettlementReconciliationGuard(inputs).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
from pathlib import Path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import SettlementReconciliationGuard, handle
def _skill_description() -> str:
text = Path(__file__).resolve().parents[1].joinpath('SKILL.md').read_text(encoding='utf-8')
frontmatter = text.split('---', 2)[1]
for line in frontmatter.splitlines():
if line.strip().startswith('description:'):
return line.split(':', 1)[1].strip().strip("\"'")
raise AssertionError('description not found in SKILL.md')
def test_review_mode_detection_month_end():
guard = SettlementReconciliationGuard('Need month-end close support for payout and refund timing review.')
assert guard.review_mode == 'Month-end close support'
assert 'Refund timing mismatch' in guard.discrepancies
def test_discrepancy_detection_missing_payout_and_fee():
guard = SettlementReconciliationGuard('Investigate missing payout, fee mismatch, and unreconciled batch differences.')
assert guard.review_mode == 'Discrepancy investigation'
assert 'Missing payout or delayed remittance' in guard.discrepancies
assert 'Fee mismatch' in guard.discrepancies
def test_channel_and_evidence_detection():
guard = SettlementReconciliationGuard('Compare Amazon payout report with bank statement and fee schedule before posting to the ERP general ledger.')
assert 'Marketplace settlement' in guard.channels
assert 'Bank payout / remittance' in guard.channels
assert 'ERP / finance ledger' in guard.channels
assert 'Payout report' in guard.evidence
assert 'Bank statement' in guard.evidence
assert 'Fee schedule' in guard.evidence
def test_render_contains_sections_and_skill_description():
output = handle('Create a weekly exception sweep for refund, reserve, and fee variance issues.')
assert output.startswith('# Settlement Reconciliation Guard')
assert f'**Skill description:** {_skill_description()}' in output
assert '## Matching Logic Table' in output
assert '## Discrepancy Queue' in output
assert '## Close Controls' in output
def test_dict_input_supported():
output = handle({
'review_mode': 'daily settlement check',
'channels': ['marketplace', 'bank'],
'evidence': ['payout report', 'bank statement', 'refund log'],
'issue': 'chargeback reserve leakage',
})
assert '# Settlement Reconciliation Guard' in output
assert 'Daily payout check' in output
assert 'Chargeback or reserve leakage' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Build a supplier sync control brief for inventory, catalog, pricing, order-handoff, and onboarding workflows. Use when a team needs to spot mapping gaps, lat...
---
name: supplier-sync-manager
description: Build a supplier sync control brief for inventory, catalog, pricing, order-handoff, and onboarding workflows. Use when a team needs to spot mapping gaps, latency risks, or supplier-data exceptions before they spread across ERP, WMS, OMS, or marketplace channels.
---
# Supplier Sync Manager
## Overview
Supplier Sync Manager turns a short operations prompt into a practical control brief for supplier-to-system synchronization work.
It is useful when you need a fast operating view of sync scope, likely failure themes, control questions, and exception ownership.
## Use this skill when
- inventory or availability updates are lagging across supplier feeds and internal systems
- catalog, SKU, pack-size, or pricing mappings look inconsistent
- order, shipment, or fulfillment events are duplicating, missing, or drifting between systems
- a new supplier onboarding, file template rollout, or cutover needs a risk checklist
- the team needs a concise exception-management brief instead of a raw troubleshooting dump
## What the skill does
The handler reads the prompt, infers likely context, and produces a structured brief with:
1. **Primary sync objective** such as inventory synchronization, catalog and pricing alignment, order handoff, onboarding, or exception review
2. **Operating cadence** such as real-time, daily, weekly, or launch preparation
3. **Systems referenced** including ERP, WMS, OMS, PIM, marketplace/storefront, or supplier portal workflows
4. **Priority risk themes** such as SKU mapping mismatch, inventory latency, price mismatch, pack-size confusion, duplicate events, or supplier SLA drift
5. **Control recommendations** covering first-check questions, default responses, watchlists, exception-queue design, and assumptions
## Recommended input patterns
Use plain language. Helpful details include:
- sync objective or pain point
- cadence or business timing
- systems involved
- known discrepancies or failure symptoms
- whether the need is daily operations, launch readiness, or post-incident review
Example prompts:
- `Need daily inventory sync between supplier feed and ERP to avoid stock lag.`
- `Review duplicate order retries and shipment tracking gaps in the OMS handoff.`
- `Prepare a launch cutover plan for catalog mapping and supplier onboarding.`
- `Our ERP, WMS, and Shopify storefront show price mismatch and pack size mapping issues.`
## Output shape
The skill returns a markdown brief with sections such as:
- Sync Scope Summary
- Recommended Control Table
- Field Mapping Watchlist
- Cadence Notes
- Exception Queue Design
- Assumptions and Limits
## Boundaries
- This skill is heuristic. It does not query live ERP, WMS, OMS, marketplace, or supplier APIs.
- It helps frame operational controls and investigation priorities, not replace production monitoring or master-data governance.
- Final reruns, field changes, SLA decisions, and supplier escalations should remain human-approved.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
OBJECTIVE_RULES = {
"Inventory synchronization": ["inventory", "stock", "availability", "qty", "quantity", "on-hand", "on hand"],
"Catalog and pricing alignment": ["catalog", "attribute", "title", "spec", "sku", "price", "cost"],
"Order and fulfillment handoff": ["order", "shipment", "tracking", "fulfillment", "asn", "po"],
"Supplier onboarding and master-data setup": ["onboarding", "new supplier", "template", "master data", "mapping", "integration"],
"Exception and discrepancy review": ["exception", "mismatch", "failed sync", "error", "duplicate", "conflict"],
}
CADENCE_RULES = {
"Real-time or intraday control": ["real-time", "realtime", "hourly", "intraday", "live"],
"Daily control cycle": ["daily", "overnight", "every day", "morning", "day-end"],
"Weekly operating review": ["weekly", "this week", "weekly review"],
"Launch or cutover preparation": ["launch", "go-live", "cutover", "first sync", "onboarding"],
}
SYSTEM_RULES = {
"ERP": ["erp", "sap", "netsuite", "kingdee", "u8"],
"WMS": ["wms", "warehouse"],
"OMS": ["oms", "order management"],
"PIM / Catalog": ["pim", "catalog", "product info"],
"Marketplace / Storefront": ["amazon", "shopify", "tiktok", "jd", "taobao", "marketplace", "storefront"],
"Supplier portal / spreadsheet": ["supplier portal", "portal", "spreadsheet", "excel", "csv", "template"],
}
RISK_RULES = {
"SKU mapping mismatch": {
"keywords": ["mapping", "sku mismatch", "wrong sku", "attribute mismatch", "id mismatch"],
"why": "Broken identifiers make every downstream sync look unstable even when the transport layer is working.",
"check": "Confirm internal SKU, supplier SKU, barcode, variant, and unit definitions resolve to one source of truth.",
"action": "Freeze risky changes, repair the mapping table, and re-run only the affected records.",
},
"Inventory latency": {
"keywords": ["inventory lag", "stock lag", "latency", "oversell", "availability", "qty"],
"why": "Late inventory updates turn healthy demand into oversells, cancellations, or false replenishment signals.",
"check": "Measure how long supplier, warehouse, and storefront stock snapshots remain inconsistent.",
"action": "Shorten the sync loop, isolate the slow system, and define a fallback availability policy.",
},
"Price or cost mismatch": {
"keywords": ["price", "cost", "margin", "currency", "discount"],
"why": "Pricing drift can create channel conflict, margin leakage, or supplier disputes very quickly.",
"check": "Review which system owns sell price, buy cost, tax, and currency conversion logic.",
"action": "Lock the price source of truth and reconcile only approved price updates across channels.",
},
"Pack size or MOQ mismatch": {
"keywords": ["pack size", "moq", "case pack", "carton", "uom", "unit"],
"why": "Unit-of-measure confusion makes purchasing, receiving, and sellable stock numbers disagree.",
"check": "Inspect whether pack, carton, MOQ, and sellable unit definitions are being converted consistently.",
"action": "Document the conversion rules and test edge cases before expanding the integration scope.",
},
"Duplicate or missing order events": {
"keywords": ["duplicate order", "missing order", "retry", "resend", "tracking", "shipment"],
"why": "Order-event drift creates fulfillment noise and corrupts downstream operational reporting.",
"check": "Review event idempotency, retry logic, and whether acknowledgements are persisted correctly.",
"action": "Quarantine duplicate flows, reissue the missing events, and tighten event-state logging.",
},
"Supplier SLA drift": {
"keywords": ["sla", "delay", "lead time", "supplier delay", "late"],
"why": "Supplier timing drift can make a technically correct sync operationally unreliable.",
"check": "Compare promised lead time, confirmation time, and exception response time against reality.",
"action": "Escalate the supplier operating review and separate data issues from true supply issues.",
},
}
class SupplierSyncManager:
def __init__(self, inputs: Any):
self.meta = _load_skill_meta()
self.raw = inputs
self.text = _normalize_inputs(inputs)
self.lower = self.text.lower()
self.objective = _detect_one(self.lower, OBJECTIVE_RULES, "Inventory synchronization")
self.cadence = _detect_one(self.lower, CADENCE_RULES, "Daily control cycle")
self.systems = _detect_many(self.lower, SYSTEM_RULES, ["ERP", "Supplier portal / spreadsheet"], limit=5)
self.risks = self._detect_risks()
def _detect_risks(self) -> List[str]:
matched = [name for name, data in RISK_RULES.items() if any(keyword in self.lower for keyword in data["keywords"])]
defaults = {
"Inventory synchronization": ["Inventory latency", "SKU mapping mismatch", "Supplier SLA drift"],
"Catalog and pricing alignment": ["Price or cost mismatch", "SKU mapping mismatch", "Pack size or MOQ mismatch"],
"Order and fulfillment handoff": ["Duplicate or missing order events", "Supplier SLA drift", "SKU mapping mismatch"],
"Supplier onboarding and master-data setup": ["SKU mapping mismatch", "Pack size or MOQ mismatch", "Supplier SLA drift"],
"Exception and discrepancy review": ["SKU mapping mismatch", "Inventory latency", "Price or cost mismatch"],
}[self.objective]
ordered: List[str] = []
for item in matched + defaults:
if item not in ordered:
ordered.append(item)
return ordered[:4]
def _control_rows(self) -> List[List[str]]:
rows: List[List[str]] = []
for risk in self.risks:
info = RISK_RULES[risk]
rows.append([risk, info["why"], info["check"], info["action"]])
return rows
def _field_watchlist(self) -> List[str]:
items = [
"Confirm one canonical key path for internal SKU, supplier SKU, barcode, and variant identifiers.",
"Review owner rules for price, cost, available quantity, lead time, and pack-size fields before automation expands.",
"Check whether nulls, blanks, and zero values have different meanings across supplier files and internal systems.",
]
if "Marketplace / Storefront" in self.systems:
items.append("Verify storefront-ready fields, publish flags, and channel-specific attribute requirements are mapped deliberately.")
if self.objective == "Order and fulfillment handoff":
items.append("Map acknowledgment, shipment, cancellation, and return statuses to one shared event language.")
return items[:4]
def _cadence_notes(self) -> List[str]:
notes = {
"Real-time or intraday control": [
"Use lightweight exception thresholds, because real-time sync work fails when every tiny delta becomes an alarm.",
"Track lag between source update and downstream visibility so teams can tell data delay from true stock risk.",
"Keep a manual fallback path for high-value SKUs when the automation is degraded.",
],
"Daily control cycle": [
"Start each cycle with the previous day's failed or partial sync queue, not just the fresh file drop.",
"Separate structural issues from one-off supplier file mistakes so the team does not overreact to noise.",
"Close the day with an owner-confirmed exception list and a re-run decision.",
],
"Weekly operating review": [
"Use the week to spot repeat breaks by supplier, field, or system rather than treating every incident as unique.",
"Review whether manual fixes are quietly replacing a missing productized control.",
"Escalate only the patterns that create real commercial or operational risk.",
],
"Launch or cutover preparation": [
"Run a pre-launch file rehearsal with edge cases, not just a happy-path sample.",
"Confirm ownership for rollback, freeze windows, and master-data approval before the first live sync.",
"Limit initial scope so the first live cycle teaches the team something controllable.",
],
}
return notes[self.cadence]
def _exception_queue(self) -> List[str]:
return [
"Critical: records that would cause oversell, wrong price, or broken order routing if left unresolved.",
"High: supplier rows missing identifiers, unit definitions, or mandatory commercial fields.",
"Medium: stale confirmations, repeated retries, or low-volume mismatches that still need root-cause ownership.",
"Owner rule: one person owns supplier follow-up and one person owns internal-system correction when the issue crosses both sides.",
]
def _assumptions(self) -> List[str]:
notes = [
"This brief is heuristic and uses only the user's context plus the local skill description.",
"No live ERP, OMS, WMS, supplier portal, or marketplace API was queried.",
"Master-data changes, supplier SLAs, and sync reruns should remain human-approved.",
]
missing = [system for system in ["WMS", "Marketplace / Storefront", "OMS"] if system not in self.systems]
if missing:
notes.append(f"Lightly referenced systems in the prompt: {_join(missing)}.")
return notes
def render(self) -> str:
lines: List[str] = []
lines.append("# Supplier Sync Management Brief")
lines.append("")
lines.append(f"**Skill description:** {self.meta['description']}")
lines.append(f"**Primary sync objective:** {self.objective}")
lines.append(f"**Operating cadence:** {self.cadence}")
lines.append(f"**Systems referenced:** {_join(self.systems)}")
lines.append(f"**Priority risk themes:** {_join(self.risks)}")
lines.append(f"**Input snapshot:** {self.text or 'No structured supplier-sync context was provided, so this brief uses default operational-control assumptions.'}")
lines.append("")
lines.append("## Sync Scope Summary")
lines.append("- Treat supplier sync as an operating-control problem, not just a file-transfer problem.")
lines.append("- Decide which system is authoritative for each commercial and operational field before scaling automation.")
lines.append(f"- Because the main objective is **{self.objective.lower()}**, the first controls should reduce repeat exceptions rather than just clean up today's queue.")
lines.append("")
lines.append("## Recommended Control Table")
lines.append("| Risk area | Why it matters | First control question | Default response |")
lines.append("|---|---|---|---|")
for row in self._control_rows():
lines.append(f"| {row[0]} | {row[1]} | {row[2]} | {row[3]} |")
lines.append("")
lines.append("## Field Mapping Watchlist")
for item in self._field_watchlist():
lines.append(f"- {item}")
lines.append("")
lines.append("## Operating Cadence")
for idx, item in enumerate(self._cadence_notes(), 1):
lines.append(f"{idx}. {item}")
lines.append("")
lines.append("## Exception Queue Design")
for item in self._exception_queue():
lines.append(f"- {item}")
lines.append("")
lines.append("## Assumptions and Limits")
for note in self._assumptions():
lines.append(f"- {note}")
return "\n".join(lines)
def handle(inputs: Any) -> str:
return SupplierSyncManager(inputs).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
from pathlib import Path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import SupplierSyncManager, handle
def _skill_description() -> str:
text = Path(__file__).resolve().parents[1].joinpath('SKILL.md').read_text(encoding='utf-8')
frontmatter = text.split('---', 2)[1]
for line in frontmatter.splitlines():
if line.strip().startswith('description:'):
return line.split(':', 1)[1].strip().strip("\"'")
raise AssertionError('description not found in SKILL.md')
def test_objective_detection_inventory():
manager = SupplierSyncManager('Need daily inventory sync between supplier feed and ERP to avoid stock lag.')
assert manager.objective == 'Inventory synchronization'
assert manager.cadence == 'Daily control cycle'
def test_objective_detection_order_handoff():
manager = SupplierSyncManager('Review duplicate order retries and shipment tracking gaps in the OMS handoff.')
assert manager.objective == 'Order and fulfillment handoff'
assert 'Duplicate or missing order events' in manager.risks
def test_system_and_risk_detection():
manager = SupplierSyncManager('Our ERP, WMS, and Shopify storefront show price mismatch and pack size mapping issues.')
assert 'ERP' in manager.systems
assert 'WMS' in manager.systems
assert 'Marketplace / Storefront' in manager.systems
assert 'Price or cost mismatch' in manager.risks
assert 'Pack size or MOQ mismatch' in manager.risks
def test_render_contains_sections_and_skill_description():
output = handle('Prepare a launch cutover plan for catalog mapping and supplier onboarding.')
assert output.startswith('# Supplier Sync Management Brief')
assert f'**Skill description:** {_skill_description()}' in output
assert '## Recommended Control Table' in output
assert '## Field Mapping Watchlist' in output
assert '## Exception Queue Design' in output
def test_dict_input_supported():
output = handle({
'objective': 'price sync',
'systems': ['ERP', 'supplier portal', 'marketplace'],
'risks': ['currency mismatch', 'mapping'],
'cadence': 'weekly',
})
assert '# Supplier Sync Management Brief' in output
assert 'Catalog and pricing alignment' in output
assert 'Weekly operating review' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Turn product briefs, pain points, competitor notes, and platform constraints into product page titles, hero benefit blocks, detail-module structure, and FAQ...
---
name: product-page-copywriter
description: Turn product briefs, pain points, competitor notes, and platform constraints into product page titles, hero benefit blocks, detail-module structure, and FAQ or closing copy. Use when operators, designers, sellers, or founders need multi-version PDP copy without live CMS, feed, or compliance integrations.
---
# Product Page Copywriter
## Overview
Use this skill to translate product parameters, user pain points, selling angles, competitor context, and platform rules into structured product page copy. It is designed for teams that need conversion-oriented copy while still respecting platform and compliance constraints.
This MVP is heuristic. It does **not** connect to live storefront CMS, marketplace feeds, compliance systems, or creative tools. It relies on the user's product brief, market context, and constraints.
## Trigger
Use this skill when the user wants to:
- write better PDP or product-detail-page copy from a rough brief
- produce multiple copy versions for different channels or tones
- structure hero claims, detail modules, FAQ, and closing copy in one pack
- separate user benefits from feature dumping
- prepare copy that designers, operators, and compliance reviewers can quickly inspect
### Example prompts
- "Write product page copy for this new kitchen gadget"
- "Create a Tmall and JD version of our PDP copy"
- "Help us rewrite our detail page around clearer user benefits"
- "Turn these product notes into a copy pack with compliance watchouts"
## Workflow
1. Capture the product, platform, target audience, and whether the main tension is conversion, tone, or compliance.
2. Extract the likely value angles, such as convenience, performance, trust, premium feel, or scenario fit.
3. Turn those angles into title options, hero selling points, module structure, and FAQ or closing copy.
4. Mark where copy may need legal or platform review.
5. Return a markdown copy pack that is practical for design and operations handoff.
## Inputs
The user can provide any mix of:
- product name, parameters, and category
- target audience and use scenarios
- pain points, benefits, or proof points
- competitor differences and platform requirements
- restricted words, compliance concerns, or sensitive claims
- preferred tone such as conversion-led, premium, concise, or cautious
## Outputs
Return a markdown copy pack with:
- title candidates
- hero selling points
- detail-module structure
- FAQ and closing copy
- compliance watchouts and assumptions
## Safety
- Do not claim access to live CMS, marketplace feeds, or compliance systems.
- Avoid absolute, medical, efficacy, or unsubstantiated comparative claims.
- Keep final legal or platform-sensitive approval human-reviewed.
- When proof is weak, downgrade bold claims into softer benefit framing.
## Best-fit Scenarios
- ecommerce PDP rewrite and launch support
- marketplace and DTC copy adaptation
- teams that need copy structure before design work starts
- operator-led businesses that need faster copy iteration with clear guardrails
## Not Ideal For
- final legal review of regulated claims
- visual design generation or image production
- product categories with no usable source material at all
- fully automated publishing into a live storefront or marketplace
## Acceptance Criteria
- Return markdown text.
- Include title, hero, structure, FAQ, and compliance sections.
- Provide at least two stylistic options or angles.
- Keep the copy pack practical for design, operations, and compliance review.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
PLATFORM_RULES = {
"Tmall": ["tmall", "天猫"],
"JD": ["jd", "jingdong", "京东"],
"DTC / Shopify": ["shopify", "dtc", "independent site", "独立站"],
"TikTok Shop": ["tiktok", "douyin", "抖音"],
"Amazon": ["amazon"],
}
TONE_RULES = {
"Compliant / cautious": ["compliance", "legal", "cautious", "conservative", "合规", "审核"],
"Conversion-led": ["conversion", "sell", "performance", "转化", "成交"],
"Brand-premium": ["premium", "brand", "high-end", "高端", "调性"],
"Marketplace concise": ["concise", "marketplace", "简洁", "平台风格"],
}
AUDIENCE_RULES = {
"Busy daily users": ["busy", "daily use", "everyday", "commute", "office", "通勤", "日常"],
"Gift buyers": ["gift", "gifting", "送礼"],
"Parents and families": ["parents", "family", "kids", "children", "家庭", "宝妈"],
"Health-conscious shoppers": ["health", "fitness", "wellness", "健康"],
"Value-driven shoppers": ["value", "budget", "save", "性价比", "省钱"],
}
ANGLE_RULES = {
"Convenience / ease": ["easy", "portable", "simple", "convenient", "lightweight", "便捷", "轻便"],
"Performance / outcome": ["performance", "effective", "powerful", "fast", "result", "效果", "高效"],
"Safety / trust": ["safe", "certified", "tested", "trusted", "warranty", "安全", "认证"],
"Style / premium feel": ["design", "premium", "sleek", "stylish", "质感", "高级"],
"Value / bundle logic": ["bundle", "value", "save", "cost", "套装", "划算"],
"Scenario fit": ["travel", "office", "home", "gym", "kitchen", "户外", "场景"],
}
RISK_RULES = {
"Absolute or exaggerated claim risk": ["best", "guarantee", "100%", "perfect", "no.1", "顶级", "绝对"],
"Efficacy or medical claim risk": ["medical", "therapy", "cure", "healing", "功效", "治疗", "改善"],
"Children or sensitive-user risk": ["baby", "child", "pregnant", "儿童", "孕妇"],
"Comparative substantiation risk": ["better than", "vs", "compare", "竞品", "领先"],
}
ANGLE_COPY = {
"Convenience / ease": "easy everyday use",
"Performance / outcome": "clear practical performance",
"Safety / trust": "trust-building proof",
"Style / premium feel": "premium product feel",
"Value / bundle logic": "value that feels justified",
"Scenario fit": "fit for real-life use moments",
}
AUDIENCE_COPY = {
"Busy daily users": "busy everyday users",
"Gift buyers": "gift-ready shoppers",
"Parents and families": "families and parents",
"Health-conscious shoppers": "health-conscious shoppers",
"Value-driven shoppers": "value-focused shoppers",
}
PLATFORM_HINT = {
"Tmall": "Tmall-ready merchandising language",
"JD": "JD-friendly clarity and spec confidence",
"DTC / Shopify": "DTC storytelling flow",
"TikTok Shop": "TikTok Shop scannable hooks",
"Amazon": "Amazon-style clarity and proof framing",
}
class ProductPageCopywriter:
def __init__(self, user_input: Any):
self.raw = user_input
self.text = _normalize_inputs(user_input)
self.lower = self.text.lower()
self.platform = _detect_one(self.text, PLATFORM_RULES, "DTC / Shopify")
self.tone = _detect_one(self.text, TONE_RULES, "Conversion-led")
self.audiences = _detect_many(self.text, AUDIENCE_RULES, ["Busy daily users", "Value-driven shoppers"], limit=3)
self.angles = _detect_many(
self.text,
ANGLE_RULES,
["Performance / outcome", "Convenience / ease", "Safety / trust"],
limit=4,
)
self.risks = _detect_many(
self.text,
RISK_RULES,
["Absolute or exaggerated claim risk", "Comparative substantiation risk"],
limit=3,
)
self.product_label = self._product_label()
def _product_label(self) -> str:
if isinstance(self.raw, dict):
for key in ["product_name", "product", "name", "sku_name", "title"]:
value = self.raw.get(key)
if value:
return str(value)
brand = self.raw.get("brand")
category = self.raw.get("category") or self.raw.get("product_type")
if brand and category:
return f"{brand} {category}"
return "[Brand/Product]"
def _title_candidates(self) -> List[str]:
audience = AUDIENCE_COPY.get(self.audiences[0], "everyday shoppers")
angle_a = ANGLE_COPY[self.angles[0]]
angle_b = ANGLE_COPY[self.angles[1]] if len(self.angles) > 1 else angle_a
angle_c = ANGLE_COPY[self.angles[2]] if len(self.angles) > 2 else angle_b
titles = [
f"Version A: {self.product_label} | {angle_a} | {PLATFORM_HINT[self.platform]}",
f"Version B: {self.product_label} for {audience} | {angle_b} without feature overload",
f"Version C: {self.product_label} with {angle_c} and a {self.tone.lower()} finish",
]
return titles
def _hero_points(self) -> List[str]:
points: List[str] = []
for angle in self.angles[:3]:
if angle == "Convenience / ease":
points.append("Lead with the simplest daily-use benefit first, so the shopper understands the time or effort saved immediately.")
elif angle == "Performance / outcome":
points.append("Translate the strongest functional result into a clear user payoff instead of repeating technical specs without context.")
elif angle == "Safety / trust":
points.append("Show the proof source early, such as testing, certification, warranty, or support confidence, before making stronger claims.")
elif angle == "Style / premium feel":
points.append("Use premium cues carefully so the page feels elevated without sounding vague or inflated.")
elif angle == "Value / bundle logic":
points.append("Frame value as smart total-use value or bundle logic, not as cheapness alone.")
else:
points.append("Anchor the copy in a concrete use scenario so the shopper can picture where the product fits in real life.")
return points
def _page_structure(self) -> List[str]:
return [
"Module 1: Headline promise plus who the product helps and why it matters now.",
"Module 2: Three proof-backed selling points ordered by user payoff, not internal spec priority.",
"Module 3: Scenario or comparison block that helps the shopper understand fit, usage, and trust.",
"Module 4: FAQ plus final close that removes the biggest hesitation before purchase.",
]
def _faq_block(self) -> List[str]:
return [
"Common concern: Is the claimed benefit clearly supported by the facts we actually have?",
"Proof to show: product specs, testing, warranty, user scenario clarity, or customer support reassurance.",
f"Closing line: Invite purchase with confidence, but keep the promise inside a {self.tone.lower()} boundary.",
]
def _compliance_notes(self) -> List[str]:
notes = [
"Replace absolute or category-leading language unless the team has documented proof.",
"Flag any medical, efficacy, or highly sensitive claims for legal or platform review before publishing.",
]
if self.platform in {"Tmall", "JD", "TikTok Shop"}:
notes.append(f"Because the platform focus is {self.platform}, keep the first screen especially scannable and policy-safe.")
for risk in self.risks:
notes.append(f"Specific watchout: {risk}.")
return notes[:5]
def render(self) -> str:
meta = _load_skill_meta()
lines: List[str] = []
lines.append("# Product Page Copy Pack")
lines.append("")
lines.append(f"**Skill description:** {meta['description']}")
lines.append(f"**Product focus:** {self.product_label}")
lines.append(f"**Platform focus:** {self.platform}")
lines.append(f"**Copy tone:** {self.tone}")
lines.append(f"**Audience cues:** {_join(self.audiences)}")
lines.append(f"**Benefit angles:** {_join(self.angles)}")
lines.append(f"**Input snapshot:** {self.text or 'No structured product brief was provided, so this pack uses default ecommerce PDP assumptions.'}")
lines.append("")
lines.append("## Title Candidates")
for title in self._title_candidates():
lines.append(f"- {title}")
lines.append("")
lines.append("## Hero Selling Points")
for idx, point in enumerate(self._hero_points(), 1):
lines.append(f"{idx}. {point}")
lines.append("")
lines.append("## Page Structure")
for item in self._page_structure():
lines.append(f"- {item}")
lines.append("")
lines.append("## FAQ / Closing")
for item in self._faq_block():
lines.append(f"- {item}")
lines.append("")
lines.append("## Compliance Watchouts")
for note in self._compliance_notes():
lines.append(f"- {note}")
return "\n".join(lines)
def handle(user_input: Any) -> str:
return ProductPageCopywriter(user_input).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import ProductPageCopywriter, handle
def test_platform_detection_jd():
writer = ProductPageCopywriter('Create JD product page copy for a new kitchen appliance.')
assert writer.platform == 'JD'
def test_tone_detection_compliance():
writer = ProductPageCopywriter('Need compliant copy with legal review because this category is sensitive.')
assert writer.tone == 'Compliant / cautious'
def test_render_contains_sections():
output = handle('Write conversion-led PDP copy for a travel bottle with portable and safe design benefits.')
assert output.startswith('# Product Page Copy Pack')
assert '## Title Candidates' in output
assert '## Compliance Watchouts' in output
def test_dict_input_supported():
output = handle({
'product_name': 'Aurora Travel Mug',
'platform': 'Tmall',
'benefits': ['portable', 'easy to clean', 'premium feel'],
})
assert '# Product Page Copy Pack' in output
assert 'Aurora Travel Mug' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Analyze ecommerce fulfillment notes across weight, volume, packaging, region, carrier pricing, and free-shipping policy, then turn rough order-cost data into...
---
name: shipping-cost-optimizer
description: Analyze ecommerce fulfillment notes across weight, volume, packaging, region, carrier pricing, and free-shipping policy, then turn rough order-cost data into a prioritized shipping cost reduction plan. Use when operators, warehouse managers, or founders need logistics cost guidance without live TMS, carrier, or ERP integrations.
---
# Shipping Cost Optimizer
## Overview
Use this skill to convert order-cost notes, packaging specs, carrier rules, and free-shipping policies into a practical logistics cost brief. It is designed for operators who need to reduce fulfillment waste without losing control of delivery experience.
This MVP is heuristic. It does **not** connect to live TMS, carrier dashboards, ERP systems, or parcel tracking feeds. It relies on the user's exported cost notes, rate summaries, and operational context.
## Trigger
Use this skill when the user wants to:
- find the biggest sources of shipping-cost waste
- compare packaging, carrier-routing, threshold, or regional-profitability options
- understand whether free-shipping policy is eroding margin too aggressively
- prepare a logistics cost review for ops, warehouse, finance, or leadership
- turn rough freight and packaging notes into a prioritized test plan
### Example prompts
- "Where are we wasting the most money on shipping?"
- "Review packaging and carrier options for our store"
- "Should we raise the free-shipping threshold?"
- "Create a shipping cost reduction brief from these order notes"
## Workflow
1. Capture the main tension, such as packaging waste, routing logic, threshold design, or regional loss-making orders.
2. Normalize the likely cost signals: weight, volume, packaging, region, carrier pricing, and surcharge behavior.
3. Separate cost reduction ideas into quick wins versus cross-functional projects.
4. Attach each idea to business impact, delivery-experience risk, and implementation difficulty.
5. Return a markdown shipping-cost report with opportunities, pilot recommendations, and limits.
## Inputs
The user can provide any mix of:
- order weight and dimensional notes
- packaging or carton specifications
- carrier pricing or routing notes
- regional mix, remote-surcharge notes, or loss-making zones
- free-shipping thresholds or promotional shipping rules
- brand, speed, damage, or unboxing constraints
## Outputs
Return a markdown report with:
- a cost snapshot
- prioritized optimization opportunities
- quick wins versus longer projects
- experience and risk notes
- assumptions and limits
## Safety
- Do not claim access to live TMS, carrier dashboards, or ERP systems.
- Keep carrier changes, policy changes, and packaging changes human-approved.
- Do not recommend cost cuts that ignore damage, speed, or premium-brand experience.
- Downgrade certainty when rate cards or regional cost data are incomplete.
## Best-fit Scenarios
- ecommerce shipping-cost reviews
- free-shipping threshold and packaging audits
- warehouse, ops, finance, and founder discussions about fulfillment margin
- businesses that need a practical first-pass logistics brief before deeper analysis
## Not Ideal For
- real-time carrier routing automation
- parcel tracking or delivery exception handling
- highly specialized cross-border compliance design
- teams with no order-cost or shipping-logic visibility at all
## Acceptance Criteria
- Return markdown text.
- Include cost snapshot, opportunity table, pilot plan, and risk notes.
- Distinguish quick wins from longer efforts.
- Keep the brief practical for ops, warehouse, and finance stakeholders.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
OBJECTIVE_RULES = {
"Packaging rationalization": ["packaging", "box", "carton", "overpack", "包材", "包装"],
"Carrier routing review": ["carrier", "courier", "routing", "快递", "承运商", "route"],
"Free-shipping policy reset": ["free shipping", "threshold", "免邮", "shipping promo"],
"Regional profitability control": ["region", "zone", "偏远", "remote", "loss-making", "区域"],
"Cross-border caution": ["cross-border", "customs", "duty", "international", "跨境"],
}
SIGNAL_RULES = {
"Order weight / volume": ["weight", "volume", "dimensional", "kg", "尺寸", "体积", "重量"],
"Packaging specs": ["packaging", "box", "carton", "包材", "包装"],
"Carrier pricing": ["carrier", "courier", "rate", "quote", "承运商", "报价", "运费"],
"Region mix / surcharge": ["region", "zone", "remote", "偏远", "区域", "surcharge"],
"Free-shipping rule": ["free shipping", "threshold", "免邮", "shipping promo"],
"Experience guardrail": ["premium", "luxury", "unboxing", "fragile", "fast delivery", "高端", "易碎", "时效"],
}
OPPORTUNITY_LIBRARY = {
"Packaging rationalization": {
"pain": "The parcel is likely paying for excess cube, filler, or unnecessary material complexity.",
"recommendation": "Standardize pack tiers, remove over-boxing where safe, and match packaging to actual order profiles instead of habit.",
"difficulty": "Low to medium",
"impact": "Reduces freight, packaging cost, and handling waste if damage risk remains controlled.",
},
"Carrier routing review": {
"pain": "The current carrier split may reflect legacy habit rather than the best current lane economics.",
"recommendation": "Re-benchmark route logic by zone, order type, and SLA so the cheapest acceptable carrier wins more often.",
"difficulty": "Medium",
"impact": "Can lower unit freight cost while protecting service-level targets when lane logic is clean.",
},
"Free-shipping policy reset": {
"pain": "The threshold may be converting orders at the cost of avoidable margin erosion.",
"recommendation": "Test a higher threshold, bundle encouragement, or member-only rule instead of subsidizing every order equally.",
"difficulty": "Medium",
"impact": "Improves order economics if conversion loss stays within an acceptable range.",
},
"Regional profitability control": {
"pain": "Some zones may look healthy on revenue but become negative after remote or low-density shipping costs.",
"recommendation": "Create region-specific rules, exclusions, or slower-service defaults where profitability repeatedly breaks.",
"difficulty": "Medium to high",
"impact": "Protects contribution margin and reveals where geographic demand is expensive to serve.",
},
"Cross-border caution": {
"pain": "Cross-border cost layers can distort shipping economics far beyond domestic assumptions.",
"recommendation": "Model duties, customs variance, and exception handling separately before rolling out any broad shipping-cost promise.",
"difficulty": "High",
"impact": "Prevents underpricing and protects against operational surprises in international lanes.",
},
}
EXPERIENCE_RULES = {
"Premium brand or unboxing sensitivity": ["premium", "luxury", "unboxing", "高端"],
"Speed-sensitive delivery promise": ["fast delivery", "same day", "next day", "时效"],
"Fragile-product protection": ["fragile", "breakable", "易碎"],
}
class ShippingCostOptimizer:
def __init__(self, user_input: Any):
self.raw = user_input
self.text = _normalize_inputs(user_input)
self.lower = self.text.lower()
self.primary_objective = _detect_one(self.text, OBJECTIVE_RULES, "Packaging rationalization")
self.signals = _detect_many(
self.text,
SIGNAL_RULES,
["Order weight / volume", "Carrier pricing", "Packaging specs"],
limit=6,
)
self.opportunities = _detect_many(
self.text,
OBJECTIVE_RULES,
["Packaging rationalization", "Carrier routing review", "Free-shipping policy reset"],
limit=4,
)
self.guardrail = _detect_one(self.text, EXPERIENCE_RULES, "Balanced shipping cost versus service level")
def _pilot_plan(self) -> List[str]:
items = [
"Weeks 1-2: measure the current waste source with one clean baseline instead of changing packaging, routing, and thresholds all at once.",
"Weeks 2-4: run the lowest-friction pilot first in one lane, region, or order profile where the economics are most obviously weak.",
"Month 2: decide whether the result is strong enough for rollout after reviewing profit, damage, speed, and customer-experience effects together.",
]
if self.primary_objective == "Free-shipping policy reset":
items[1] = "Weeks 2-4: test the revised threshold or offer rule in one traffic cohort so conversion and margin quality can be read together."
if self.primary_objective == "Carrier routing review":
items[1] = "Weeks 2-4: pilot a cleaner carrier split in one stable lane before rewriting the whole routing policy."
return items
def _risk_notes(self) -> List[str]:
notes = [
"Cost reduction is only good if delivery speed, damage rate, and service quality remain inside acceptable guardrails.",
"Keep carrier contract changes, routing rules, and shipping-policy updates human-approved.",
]
if self.guardrail != "Balanced shipping cost versus service level":
notes.append(f"Special watchout: {self.guardrail} should be protected during any pilot.")
if "Cross-border caution" in self.opportunities:
notes.append("International shipping assumptions often drift quickly, so broad cross-border promises should be treated carefully.")
return notes
def render(self) -> str:
meta = _load_skill_meta()
lines: List[str] = []
lines.append("# Shipping Cost Optimization Report")
lines.append("")
lines.append(f"**Skill description:** {meta['description']}")
lines.append(f"**Primary objective:** {self.primary_objective}")
lines.append(f"**Signals referenced:** {_join(self.signals)}")
lines.append(f"**Experience guardrail:** {self.guardrail}")
lines.append(f"**Input snapshot:** {self.text or 'No structured logistics brief was provided, so this report uses default ecommerce fulfillment assumptions.'}")
lines.append("")
lines.append("## Cost Snapshot")
lines.append("- Start with the largest repeatable waste source before chasing every marginal shipping-cost hypothesis at once.")
lines.append(f"- Because the main job is **{self.primary_objective.lower()}**, the first pilot should isolate that lever and avoid mixing too many policy changes together.")
lines.append("- Treat shipping cost as a combined packaging, carrier, policy, and customer-experience problem, not just a freight invoice line.")
lines.append("")
lines.append("## Optimization Opportunities")
lines.append("| Opportunity | Current pain pattern | Recommendation | Difficulty | Business impact |")
lines.append("|---|---|---|---|---|")
for opportunity in self.opportunities:
details = OPPORTUNITY_LIBRARY[opportunity]
lines.append(
f"| {opportunity} | {details['pain']} | {details['recommendation']} | {details['difficulty']} | {details['impact']} |"
)
lines.append("")
lines.append("## Pilot Recommendations")
for idx, item in enumerate(self._pilot_plan(), 1):
lines.append(f"{idx}. {item}")
lines.append("")
lines.append("## Experience and Risk Notes")
for note in self._risk_notes():
lines.append(f"- {note}")
lines.append("")
lines.append("## Assumptions and Limits")
lines.append("- This is a heuristic cost-reduction brief, not a live TMS or carrier-routing engine.")
lines.append("- Incomplete rate cards or regional data will reduce confidence in the projected savings range.")
lines.append("- Final policy, routing, packaging, and customer-facing promise changes should stay human-approved.")
return "\n".join(lines)
def handle(user_input: Any) -> str:
return ShippingCostOptimizer(user_input).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import ShippingCostOptimizer, handle
def test_primary_objective_detection_packaging():
optimizer = ShippingCostOptimizer('Review our packaging waste, over-boxing, and carton sizes for ecommerce orders.')
assert optimizer.primary_objective == 'Packaging rationalization'
def test_signal_detection_weight_and_region():
optimizer = ShippingCostOptimizer('We need to analyze parcel weight, volumetric pricing, remote region surcharges, and carrier quotes.')
assert 'Order weight / volume' in optimizer.signals
assert 'Region mix / surcharge' in optimizer.signals
assert 'Carrier pricing' in optimizer.signals
def test_render_contains_sections():
output = handle('Should we raise the free-shipping threshold and review carrier routing?')
assert output.startswith('# Shipping Cost Optimization Report')
assert '## Optimization Opportunities' in output
assert '## Pilot Recommendations' in output
def test_dict_input_supported():
output = handle({
'weights': '0.8kg average',
'carrier': 'SF and YTO',
'policy': 'free shipping above 59',
})
assert '# Shipping Cost Optimization Report' in output
assert 'free shipping above 59' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Design monthly, quarterly, or annual ecommerce promotion calendars using business goals, platform mega-sale windows, inventory posture, and team capacity, th...
---
name: promotion-calendar-planner
description: Design monthly, quarterly, or annual ecommerce promotion calendars using business goals, platform mega-sale windows, inventory posture, and team capacity, then turn rough planning notes into a sequenced promotion roadmap. Use when operators, marketers, or managers need a promotion cadence plan without live calendar, ad-platform, or project-management integrations.
---
# Promotion Calendar Planner
## Overview
Use this skill to turn annual goals, platform nodes, inventory posture, and team resource notes into a structured promotion calendar. It is designed for operators who need a promotion cadence they can explain to management and hand off to cross-functional teams.
This MVP is heuristic. It does **not** connect to live calendars, ad platforms, merchandising systems, or project-management tools. It relies on the user's planning notes, seasonal context, and team constraints.
## Trigger
Use this skill when the user wants to:
- map a monthly, quarterly, or annual promotion cadence
- decide which sale nodes deserve a full push versus a lighter supporting role
- balance revenue, margin, inventory digestion, and team workload in one schedule
- prepare a promotion roadmap for ops, design, supply chain, and customer service
- turn rough campaign ideas into a sequenced business calendar with risk notes
### Example prompts
- "Build our Q4 promotion calendar around 11.11 and Black Friday"
- "Create a quarterly promo plan that also helps us digest aged inventory"
- "We need a yearly promotion roadmap for our ecommerce team"
- "Plan a lighter monthly cadence without overloading design and customer service"
## Workflow
1. Capture the planning window, business goal, seasonal anchors, and major capacity constraints.
2. Separate mandatory nodes, optional support nodes, and nodes that should probably be skipped.
3. Translate each node into a role such as acquisition, conversion, clearance, or retention.
4. Attach category focus, price posture, team preparation needs, and avoidable conflicts.
5. Return a markdown promotion plan with a calendar table, cadence principles, and a team checklist.
## Inputs
The user can provide any mix of:
- annual, quarterly, or monthly business goals
- platform nodes such as 618, Prime Day, 11.11, Black Friday, or brand days
- inventory posture such as healthy core stock, aged stock, or constrained hero SKUs
- category seasonality and hero categories
- resource limits across ops, design, supply chain, CRM, media, or customer service
- margin rules, discount rules, or customer-experience constraints
## Outputs
Return a markdown plan with:
- a promotion calendar table
- cadence principles before, during, and after major nodes
- a team preparation checklist
- conflict watchouts and risk notes
- assumptions and limits
## Safety
- Do not claim access to live calendars, platform policy feeds, or ad platforms.
- Verify official sale dates and platform rules before locking any execution plan.
- Keep pricing, discount depth, and activation decisions human-approved.
- Treat the plan as a baseline operating rhythm, not a substitute for daily campaign management.
## Best-fit Scenarios
- quarterly and annual ecommerce promotion planning
- cross-functional promotion alignment across ops, design, supply chain, and service
- businesses that need more structure than a spreadsheet of holiday names
- teams trying to reduce promotion chaos and discount overlap
## Not Ideal For
- live campaign pacing and day-to-day media management
- fully automated marketing orchestration
- businesses with no meaningful seasonality or promotion planning rhythm
- regulated categories where promotional language requires extensive review first
## Acceptance Criteria
- Return markdown text.
- Include calendar table, cadence principles, team checklist, and risk notes.
- Explain why key nodes matter instead of listing dates only.
- Surface at least two avoidable planning conflicts.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
WINDOW_RULES = {
"Annual plan": ["annual", "year plan", "full year", "全年"],
"Q1": ["q1", "first quarter"],
"Q2": ["q2", "second quarter"],
"Q3": ["q3", "third quarter"],
"Q4": ["q4", "fourth quarter", "holiday season"],
"Quarterly plan": ["quarter", "quarterly"],
"Monthly sprint": ["month", "monthly", "30-day", "30 day", "4 weeks"],
}
GOAL_RULES = {
"Inventory digestion": ["clearance", "aged stock", "inventory", "sell-through", "去库存", "清库存"],
"Profit discipline": ["profit", "margin", "gross margin", "毛利", "利润"],
"New-customer acquisition": ["new customer", "acquisition", "new users", "拉新"],
"Member / retention": ["member", "vip", "retention", "repeat", "会员", "复购"],
"Revenue growth": ["revenue", "gmv", "sales", "top line", "增长", "销量"],
}
ANCHOR_RULES = {
"618": ["618"],
"Prime Day": ["prime day"],
"11.11": ["11.11", "singles day", "双11"],
"Black Friday": ["black friday", "bfcm"],
"Lunar New Year": ["lunar new year", "spring festival", "春节"],
"Back-to-school": ["back to school"],
"Summer push": ["summer", "暑期"],
"Brand day": ["brand day", "anniversary", "店庆"],
}
RESOURCE_RULES = {
"Operations": ["ops", "运营", "operations"],
"Design": ["design", "creative", "设计"],
"Supply chain": ["supply chain", "inventory", "procurement", "供应链", "备货"],
"Customer service": ["customer service", "support", "客服"],
"CRM / lifecycle": ["crm", "member", "email", "私域", "会员"],
"Media / traffic": ["ads", "media", "traffic", "投放"],
}
HERO_CATEGORY_MAP = {
"Revenue growth": "Hero conversion SKUs and proven best sellers",
"Profit discipline": "High-margin core assortment and protected bundles",
"Inventory digestion": "Aged stock, bundle candidates, and low-risk clearance items",
"New-customer acquisition": "Entry-price products, trial bundles, and first-order magnets",
"Member / retention": "VIP favorites, replenishment products, and member-only sets",
}
PRICE_STRATEGY_MAP = {
"Revenue growth": "Use the strongest commercially proven offer, but keep one margin guardrail visible.",
"Profit discipline": "Prefer moderate discounts, bundles, thresholds, or gifts instead of blanket markdowns.",
"Inventory digestion": "Use planned clearance lanes so aged stock does not contaminate the full assortment narrative.",
"New-customer acquisition": "Use a clear first-order hook without training the whole customer base to wait for deep discounts.",
"Member / retention": "Use exclusivity, early access, or loyalty value before defaulting to the deepest public promotion.",
}
def _default_anchors(window: str) -> List[str]:
defaults = {
"Annual plan": ["Lunar New Year", "618", "11.11", "Black Friday"],
"Q1": ["Lunar New Year", "Brand day", "Spring refresh"],
"Q2": ["618", "Summer push", "Brand day"],
"Q3": ["Back-to-school", "Summer push", "Brand day"],
"Q4": ["11.11", "Black Friday", "Holiday gifting"],
"Quarterly plan": ["Primary sale node", "Support node", "Retention node"],
"Monthly sprint": ["Teaser week", "Main push", "Recovery week"],
}
return defaults.get(window, ["Primary sale node", "Support node", "Retention node"])
def _resource_defaults(goal: str) -> List[str]:
base = ["Operations", "Design", "Supply chain", "Customer service"]
if goal in {"Member / retention", "New-customer acquisition"}:
base.append("CRM / lifecycle")
if goal in {"Revenue growth", "New-customer acquisition"}:
base.append("Media / traffic")
return base[:5]
class PromotionCalendarPlanner:
def __init__(self, user_input: Any):
self.raw = user_input
self.text = _normalize_inputs(user_input)
self.lower = self.text.lower()
self.window = _detect_one(self.text, WINDOW_RULES, "Quarterly plan")
self.goal = _detect_one(self.text, GOAL_RULES, "Revenue growth")
self.anchors = _detect_many(self.text, ANCHOR_RULES, _default_anchors(self.window), limit=4)
self.resources = _detect_many(self.text, RESOURCE_RULES, _resource_defaults(self.goal), limit=5)
def _calendar_rows(self) -> List[str]:
roles = [
"Demand preheat / positioning",
"Main conversion push",
"Bridge or follow-up wave",
"Retention or cleanup wave",
]
rows: List[str] = []
for idx, anchor in enumerate(self.anchors):
role = roles[idx] if idx < len(roles) else "Support wave"
risk_note = "Watch workload concentration and discount overlap." if idx < 2 else "Check whether the follow-up wave still serves a distinct purpose."
rows.append(
f"| {anchor} | {role} | {HERO_CATEGORY_MAP[self.goal]} | {PRICE_STRATEGY_MAP[self.goal]} | {_join(self.resources[:3])} | {risk_note} |"
)
return rows
def _cadence_principles(self) -> List[str]:
return [
"Before the major node, align hero categories, offer logic, and stock confidence before creative production scales.",
"During the peak, keep one clear job for each node so acquisition, conversion, clearance, and retention are not all competing in the same week.",
"After the peak, plan a lighter recovery or retention beat so customer service, fulfillment, and CRM do not collapse into reactive firefighting.",
]
def _team_checklist(self) -> List[str]:
return [
"Operations: confirm channel readiness, coupon rules, and the node-by-node approval owner.",
"Design: prepare hero assets early and avoid stacking every revision into the last week before launch.",
"Supply chain: verify hero SKU availability, aged-stock lanes, and whether promo promises match actual inventory reality.",
"Customer service: prepare scripts for offer clarity, delay handling, and predictable FAQ spikes.",
"CRM / lifecycle: pre-build reminder, follow-up, and retention touchpoints for the highest-priority nodes.",
]
def _conflict_watchlist(self) -> List[str]:
conflicts = [
"Do not run two major promotions so close together that design, support, and inventory teams cannot reset cleanly.",
"Do not let a clearance wave and a premium hero launch share the same story unless the assortment split is explicit.",
]
if self.goal == "Profit discipline":
conflicts.append("Avoid chasing GMV at the cost of margin by repeating deep discounts without a profitability review.")
elif self.goal == "Inventory digestion":
conflicts.append("Do not let aged-stock clearance consume the entire calendar if hero SKUs still need a healthier brand narrative.")
else:
conflicts.append("Make sure acquisition-heavy nodes still have a retention or follow-up path so value does not end with the first order.")
return conflicts
def _assumptions(self) -> List[str]:
return [
"Official platform sale dates, policy changes, and merchandising rules should be re-verified before execution.",
"This plan is a baseline operating rhythm, not a live campaign pacing dashboard.",
"Discount depth, traffic budget, and launch approvals should remain human-approved.",
]
def render(self) -> str:
meta = _load_skill_meta()
lines: List[str] = []
lines.append("# Promotion Calendar Plan")
lines.append("")
lines.append(f"**Skill description:** {meta['description']}")
lines.append(f"**Planning window:** {self.window}")
lines.append(f"**Primary commercial goal:** {self.goal}")
lines.append(f"**Seasonal anchors:** {_join(self.anchors)}")
lines.append(f"**Cross-functional focus:** {_join(self.resources)}")
lines.append(f"**Input snapshot:** {self.text or 'No structured planning brief was provided, so this plan uses default ecommerce promotion-planning assumptions.'}")
lines.append("")
lines.append("## Calendar Recommendations")
lines.append("| Month / node | Activity role | Hero category | Price strategy | Resource need | Risk note |")
lines.append("|---|---|---|---|---|---|")
lines.extend(self._calendar_rows())
lines.append("")
lines.append("## Cadence Principles")
for bullet in self._cadence_principles():
lines.append(f"- {bullet}")
lines.append("")
lines.append("## Team Preparation Checklist")
for item in self._team_checklist():
lines.append(f"- {item}")
lines.append("")
lines.append("## Conflict Watchlist")
for idx, item in enumerate(self._conflict_watchlist(), 1):
lines.append(f"{idx}. {item}")
lines.append("")
lines.append("## Assumptions and Limits")
for note in self._assumptions():
lines.append(f"- {note}")
return "\n".join(lines)
def handle(user_input: Any) -> str:
return PromotionCalendarPlanner(user_input).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import PromotionCalendarPlanner, handle
def test_window_detection_q4():
planner = PromotionCalendarPlanner('Build our Q4 promotion calendar around 11.11 and Black Friday.')
assert planner.window == 'Q4'
assert '11.11' in planner.anchors or 'Black Friday' in planner.anchors
def test_goal_detection_inventory_digestion():
planner = PromotionCalendarPlanner('Create a quarterly promo plan to clear aged inventory and improve sell-through.')
assert planner.goal == 'Inventory digestion'
def test_render_contains_sections():
output = handle('Plan a quarterly promotion calendar for revenue growth with 618 as the main anchor.')
assert output.startswith('# Promotion Calendar Plan')
assert '## Calendar Recommendations' in output
assert '## Team Preparation Checklist' in output
assert '## Conflict Watchlist' in output
def test_dict_input_supported():
output = handle({
'window': 'annual',
'goal': 'margin and member retention',
'anchors': ['Lunar New Year', '618', '11.11'],
})
assert '# Promotion Calendar Plan' in output
assert 'Lunar New Year' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Segment ecommerce customers by repeat behavior, margin quality, membership depth, and churn or return risk, then turn rough order-history notes into a priori...
---
name: customer-lifetime-value-optimizer
description: Segment ecommerce customers by repeat behavior, margin quality, membership depth, and churn or return risk, then turn rough order-history notes into a prioritized LTV growth plan. Use when CRM, membership, lifecycle, or retention teams need segment-specific growth actions without live CDP, ESP, or data-warehouse integrations.
---
# Customer Lifetime Value Optimizer
## Overview
Use this skill to convert customer-segment notes, order-history summaries, gross-margin signals, and retention context into a practical LTV action plan. It is built for operators who need fast prioritization across new-customer nurture, repeat purchase growth, margin protection, and winback strategy.
This MVP is heuristic. It does **not** connect to live CRM, CDP, ESP, loyalty, or analytics systems. It relies on the user's segment notes, exported summaries, and lifecycle context.
## Trigger
Use this skill when the user wants to:
- identify which customer segments deserve the most retention investment
- design different lifecycle moves for high-value, price-sensitive, dormant, or return-risk customers
- rank LTV levers such as repeat rate, AOV, margin mix, or churn reduction
- turn rough order-history notes into a CRM or membership action brief
- separate revenue growth ideas from margin-quality and retention-quality risks
### Example prompts
- "Which segments should we prioritize to improve LTV this quarter?"
- "Create a retention plan for VIP, new, and dormant customers"
- "How can we grow LTV without overusing discounts?"
- "Turn these order and membership notes into an LTV roadmap"
## Workflow
1. Capture the customer segments, order behavior, and whether the main tension is repeat rate, AOV, churn, or margin quality.
2. Normalize the likely LTV signals: order history, repurchase cycle, segment mix, return behavior, and offer sensitivity.
3. Separate customer groups into different action lanes instead of giving one generic lifecycle answer.
4. Rank the highest-value LTV levers and attach practical plays, owners, and success metrics.
5. Return a markdown plan with segment diagnosis, lever ranking, and action packages.
## Inputs
The user can provide any mix of:
- customer segments or membership tiers
- order history and repeat-cycle notes
- AOV, gross margin, bundle rate, or attach-rate context
- churn, dormancy, or lapsed-customer notes
- refund or return-risk observations
- lifecycle messaging constraints and incentive constraints
## Outputs
Return a markdown plan with:
- a segment diagnosis table
- ranked LTV levers
- action packages by segment
- short, medium, and longer-horizon priorities
- measurement notes, assumptions, and limits
## Safety
- Do not claim access to live CRM, ESP, loyalty, or analytics systems.
- Do not auto-send discounts, coupons, or lifecycle messages.
- Keep revenue lift and margin impact separate in the recommendations.
- Downgrade certainty when user-level order history is incomplete.
- Treat financial LTV models and operator-facing lifecycle plans as related but not identical.
## Best-fit Scenarios
- CRM and membership planning for ecommerce teams
- repeat-purchase and lifecycle improvement reviews
- retention strategy design when data is partial but usable
- operator-led businesses that need an action plan before building a deeper model
## Not Ideal For
- formal finance-grade LTV forecasting
- automatic customer scoring or trigger orchestration
- businesses with no segment or order-history visibility at all
- scenarios that require privacy-reviewed activation logic
## Acceptance Criteria
- Return markdown text.
- Include segment diagnosis, lever ranking, action packages, and limits.
- Show at least one short-term, one medium-term, and one longer-term move.
- Keep the plan practical for CRM, lifecycle, and retention operators.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
SEGMENT_RULES = {
"High-potential new customers": ["new customer", "new buyer", "first order", "first-time", "新客", "首购"],
"High-value loyal customers": ["vip", "loyal", "repeat", "member", "high value", "老客", "会员"],
"Price-sensitive customers": ["discount", "coupon", "deal seeker", "price sensitive", "促销", "价格敏感"],
"Dormant customers": ["dormant", "lapsed", "inactive", "sleeping", "winback", "沉睡客", "流失"],
"Return-risk customers": ["return", "refund", "exchange", "complaint", "退货", "退款"],
}
LEVER_RULES = {
"Lift repeat rate": ["repeat", "frequency", "reorder", "refill", "repurchase", "复购"],
"Raise AOV": ["aov", "average order", "bundle", "upsell", "cross sell", "加价购", "连带"],
"Protect gross margin": ["margin", "profit", "gross margin", "折扣", "discount abuse", "毛利"],
"Reduce churn": ["churn", "lapsed", "dormant", "inactive", "drop-off", "流失", "召回"],
"Improve first-order experience": ["first order", "welcome", "onboarding", "首购", "首次体验", "post-purchase"],
}
CONTEXT_RULES = {
"Low-frequency / high-ticket": ["high ticket", "big ticket", "low frequency", "furniture", "appliance", "大件"],
"Consumables / repeatable": ["consumable", "subscription", "refill", "repeat", "supplement", "复购"],
"Discount-led acquisition": ["discount", "coupon", "deal", "promotion only", "低价流量", "价格战"],
}
SEGMENT_LIBRARY = {
"High-potential new customers": {
"pattern": "First-order intent exists, but the second-order path is not yet protected.",
"issue": "The business may be paying acquisition cost without converting it into durable behavior.",
"move": "Use a tighter welcome-to-second-order path with one clear reason to come back quickly.",
"touchpoint": "Welcome flow, post-purchase education, and a timed second-order reminder.",
"offer": "Keep incentives narrow and tied to next-best products or replenishment logic.",
"metric": "30-day repeat rate and second-order contribution margin.",
},
"High-value loyal customers": {
"pattern": "These customers already compound value through repeat behavior, trust, or membership depth.",
"issue": "Generic campaigns can under-serve them or train them to wait for unnecessary discounts.",
"move": "Protect them with differentiated value, early access, or curated higher-margin bundles.",
"touchpoint": "VIP CRM, loyalty tiers, priority launches, and service recovery priority.",
"offer": "Use exclusivity, premium bundles, or convenience benefits before defaulting to deep discounts.",
"metric": "90-day gross profit contribution, repeat frequency, and premium mix.",
},
"Price-sensitive customers": {
"pattern": "They respond to offers, but their margin quality can collapse if every touch is price-led.",
"issue": "Revenue may look healthy while net value is thin or unstable.",
"move": "Shift from blanket discounts to threshold mechanics, bundles, or product selection control.",
"touchpoint": "Promo segmentation, threshold offers, and margin-protected assortment selection.",
"offer": "Prefer bundle logic, add-on logic, or category fences instead of storewide discounting.",
"metric": "Offer redemption quality, AOV, and contribution margin after discount.",
},
"Dormant customers": {
"pattern": "Past purchase intent exists, but relevance or timing has faded.",
"issue": "Winback costs rise fast if the team waits too long or uses the same message for everyone.",
"move": "Use reason-based reactivation with tailored hooks such as refill timing, newness, or solved pain points.",
"touchpoint": "Winback journeys, segmented reminders, and category-specific comeback offers.",
"offer": "Use the lightest offer that reopens attention, then graduate customers into a healthier repeat path.",
"metric": "Reactivation rate, 60-day retained value, and margin after winback spend.",
},
"Return-risk customers": {
"pattern": "Top-line spend can look attractive while refunds or dissatisfaction erase the real value.",
"issue": "The business may over-invest in segments that produce noisy or low-net revenue.",
"move": "Fix expectation setting, product matching, and onboarding before scaling lifecycle pressure.",
"touchpoint": "Pre-purchase education, FAQ clarity, order follow-up, and support escalation rules.",
"offer": "Prioritize trust-building and expectation alignment over harder promotional pushes.",
"metric": "Net revenue after returns, return rate, and complaint rate.",
},
}
LEVER_LIBRARY = {
"Lift repeat rate": "Tighten the time-to-second-order path and give the best-fit segments a clear reason to return before attention decays.",
"Raise AOV": "Use bundles, attach logic, or premium alternatives so extra revenue also improves order quality.",
"Protect gross margin": "Separate growth that looks good on revenue from growth that still creates durable profit after incentives and service cost.",
"Reduce churn": "Intervene earlier around dormancy and post-purchase drop-off instead of waiting for full customer loss.",
"Improve first-order experience": "Use onboarding, expectation setting, and early service touchpoints so more first orders become repeat behavior.",
}
class CustomerLifetimeValueOptimizer:
def __init__(self, user_input: Any):
self.raw = user_input
self.text = _normalize_inputs(user_input)
self.lower = self.text.lower()
self.segments = _detect_many(
self.text,
SEGMENT_RULES,
["High-potential new customers", "High-value loyal customers", "Dormant customers"],
limit=4,
)
self.primary_lever = _detect_one(self.text, LEVER_RULES, "Lift repeat rate")
self.levers = _detect_many(
self.text,
LEVER_RULES,
["Lift repeat rate", "Raise AOV", "Reduce churn"],
limit=4,
)
self.context = _detect_one(self.text, CONTEXT_RULES, "General ecommerce lifecycle")
def _segment_rows(self) -> List[str]:
rows: List[str] = []
for segment in self.segments:
details = SEGMENT_LIBRARY[segment]
rows.append(
f"| {segment} | {details['pattern']} | {details['issue']} | {details['move']} |"
)
return rows
def _action_packages(self) -> List[str]:
blocks: List[str] = []
for segment in self.segments[:3]:
details = SEGMENT_LIBRARY[segment]
blocks.append(f"### {segment}")
blocks.append(f"- Goal: {details['move']}")
blocks.append(f"- Touchpoints: {details['touchpoint']}")
blocks.append(f"- Offer or content logic: {details['offer']}")
blocks.append(f"- Success metric: {details['metric']}")
blocks.append("")
return blocks
def _measurement_notes(self) -> List[str]:
notes = [
"Treat this as an operator-facing LTV action brief, not a finance-grade discounted cash flow model.",
"Separate revenue lift from gross margin impact and return-adjusted value before scaling any segment play.",
"Coupon, loyalty, and CRM activation should remain human-approved.",
]
if self.context == "Low-frequency / high-ticket":
notes.append("Because the business context looks low-frequency or high-ticket, favor trust, timing, and service quality over forced repeat-rate logic.")
if self.context == "Discount-led acquisition":
notes.append("Because discount sensitivity appears material, audit incentive quality before mistaking promo response for durable LTV improvement.")
if "Return-risk customers" not in self.segments:
notes.append("If return or refund behavior is meaningful, add a separate net-value lens before treating top-line spend as real lifetime value.")
return notes
def render(self) -> str:
meta = _load_skill_meta()
lines: List[str] = []
lines.append("# LTV Optimization Plan")
lines.append("")
lines.append(f"**Skill description:** {meta['description']}")
lines.append(f"**Primary growth angle:** {self.primary_lever}")
lines.append(f"**Segments referenced:** {_join(self.segments)}")
lines.append(f"**Operating context:** {self.context}")
lines.append(f"**Input snapshot:** {self.text or 'No structured customer data was provided, so this plan uses default ecommerce lifecycle assumptions.'}")
lines.append("")
lines.append("## Segment Diagnosis")
lines.append("| Segment | Current value pattern | Main issue | Best next move |")
lines.append("|---|---|---|---|")
lines.extend(self._segment_rows())
lines.append("")
lines.append("## Lever Ranking")
for idx, lever in enumerate(self.levers, 1):
lines.append(f"{idx}. **{lever}:** {LEVER_LIBRARY[lever]}")
lines.append("")
lines.append("## Action Packages")
lines.extend(self._action_packages())
lines.append("## Measurement and Limits")
for note in self._measurement_notes():
lines.append(f"- {note}")
return "\n".join(lines)
def handle(user_input: Any) -> str:
return CustomerLifetimeValueOptimizer(user_input).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import CustomerLifetimeValueOptimizer, handle
def test_segment_detection_vip_and_dormant():
optimizer = CustomerLifetimeValueOptimizer('We need a plan for VIP members and dormant customers who stopped ordering.')
assert 'High-value loyal customers' in optimizer.segments
assert 'Dormant customers' in optimizer.segments
def test_primary_lever_detection_raise_aov():
optimizer = CustomerLifetimeValueOptimizer('Need to improve bundle attach rate, upsell, and average order value without hurting margin.')
assert optimizer.primary_lever == 'Raise AOV'
def test_render_contains_sections():
output = handle('Build an LTV plan for new customers, VIP members, and price-sensitive customers.')
assert output.startswith('# LTV Optimization Plan')
assert '## Segment Diagnosis' in output
assert '## Action Packages' in output
def test_dict_input_supported():
output = handle({
'segments': ['new customers', 'VIP members', 'dormant customers'],
'focus': 'repeat rate and margin',
'constraints': 'avoid over-discounting',
})
assert '# LTV Optimization Plan' in output
assert 'avoid over-discounting' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Audit ecommerce inventory notes or SKU exports, flag stockout, aging, overstock, capital lock-up, and promo-readiness risks, then turn partial inventory data...
---
name: inventory-health-auditor
description: Audit ecommerce inventory notes or SKU exports, flag stockout, aging, overstock, capital lock-up, and promo-readiness risks, then turn partial inventory data into a prioritized replenish-hold-clear brief. Use when operators, buyers, planners, or founders need a weekly or monthly inventory diagnosis without live ERP, WMS, or BI integrations.
---
# Inventory Health Auditor
## Overview
Use this skill to translate SKU inventory notes, sales summaries, replenishment lead times, and campaign plans into a practical inventory action report. It is designed for weekly ops reviews, monthly inventory meetings, and pre-promo readiness checks.
This MVP is heuristic. It does **not** connect to live ERP, WMS, procurement, or marketplace systems. It relies on the user's exported data, pasted notes, and business context.
## Trigger
Use this skill when the user wants to:
- identify which SKUs are most at risk of stockout, aging, or overstock
- prioritize replenishment versus hold versus clearance actions
- prepare inventory review notes for ops, procurement, or warehouse meetings
- assess whether current stock is ready for a promotion or seasonal demand spike
- turn partial inventory exports into a management summary plus execution checklist
### Example prompts
- "Review our SKU inventory and tell me what needs replenishment first"
- "Which products are aging and tying up too much cash?"
- "Prepare an inventory health brief before our 618 promotion"
- "Help me spot stockout risk across our hero SKUs"
## Workflow
1. Capture the review goal, SKU scope, and whether the immediate concern is stockout, aging, or promo readiness.
2. Normalize the main signals: on-hand inventory, sales velocity, lead time, campaign timing, and seasonality.
3. Separate the likely risk pools: stockout risk, aging or overstock risk, capital lock-up, and structural imbalance.
4. Turn each risk pool into concrete actions such as replenish, throttle, bundle, clear, or pause purchasing.
5. Return a markdown inventory report with a priority queue and a 30-day action plan.
## Inputs
The user can provide any mix of:
- SKU inventory or on-hand stock notes
- 30 to 90 day sales velocity or sell-through context
- replenishment lead times, supplier constraints, or arrival timing
- campaign plans, promo windows, or seasonal tags
- category labels, hero SKUs, and long-tail notes
- cash, warehouse, or procurement constraints
## Outputs
Return a markdown report with:
- an inventory health summary
- a priority queue of major SKU risk types
- recommended actions across replenish, hold, clear, and review
- a 30-day action plan
- assumptions, evidence gaps, and limits
## Safety
- Do not claim access to live ERP, WMS, procurement, or forecasting systems.
- Keep replenishment quantities and purchase decisions human-approved.
- Downgrade certainty when sales velocity, lead time, or campaign timing is incomplete.
- Do not treat seasonality or new-product demand as proven when historical data is weak.
## Best-fit Scenarios
- weekly or monthly ecommerce inventory reviews
- pre-promo inventory readiness checks
- multi-SKU catalog management without heavy BI tooling
- operator or founder-led businesses that need a fast prioritization layer
## Not Ideal For
- real-time replenishment automation
- warehouse slotting optimization
- full demand forecasting or probabilistic planning models
- businesses with no SKU-level inventory or sales visibility at all
## Acceptance Criteria
- Return markdown text.
- Include summary, priority queue, action plan, and limits.
- Make the advisory framing explicit.
- Keep the report practical for ops, procurement, and inventory owners.
FILE:handler.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
def _load_skill_meta() -> Dict[str, str]:
skill_path = Path(__file__).with_name("SKILL.md")
text = skill_path.read_text(encoding="utf-8")
frontmatter = ""
body = text
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
frontmatter = parts[1]
body = parts[2]
data: Dict[str, str] = {}
for line in frontmatter.splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
data[key.strip()] = value.strip().strip("\"'")
title = data.get("name", Path(__file__).resolve().parent.name.replace("-", " ").title())
for line in body.splitlines():
if line.startswith("# "):
title = line[2:].strip()
break
return {
"name": data.get("name", title),
"title": title,
"description": data.get("description", title),
}
def _normalize_inputs(inputs: Any) -> str:
if inputs is None:
return ""
if isinstance(inputs, str):
return inputs.strip()
if isinstance(inputs, dict):
parts: List[str] = []
for key, value in inputs.items():
if value in (None, "", [], {}, ()): # type: ignore[comparison-overlap]
continue
if isinstance(value, (list, tuple, set)):
rendered = ", ".join(str(item) for item in value)
else:
rendered = str(value)
parts.append(f"{key}: {rendered}")
return " | ".join(parts)
if isinstance(inputs, (list, tuple, set)):
return " | ".join(str(item) for item in inputs)
try:
return json.dumps(inputs, ensure_ascii=False, sort_keys=True)
except TypeError:
return str(inputs)
def _detect_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
lower = text.lower()
found = [label for label, keywords in rules.items() if any(keyword in lower for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _detect_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
lower = text.lower()
for label, keywords in rules.items():
if any(keyword in lower for keyword in keywords):
return label
return default
def _join(items: Sequence[str]) -> str:
return ", ".join(items) if items else "Not specified"
OBJECTIVE_RULES = {
"Stockout prevention": ["stockout", "out of stock", "缺货", "断货", "replenish", "补货", "hero sku", "hot seller", "fast seller"],
"Aging cleanup": ["aged", "aging", "滞销", "dead stock", "clearance", "slow mover", "slow-moving", "overstock"],
"Promo readiness": ["promo", "campaign", "活动", "大促", "618", "11.11", "launch", "peak season", "seasonal push"],
"Cash discipline": ["cash", "working capital", "资金", "inventory amount", "holding cost", "capital"],
}
SIGNAL_RULES = {
"Inventory on hand": ["inventory", "stock", "库存", "on hand", "units", "qty"],
"Sales velocity": ["sales", "销量", "velocity", "sell-through", "sell through", "gmv", "orders"],
"Lead time / replenishment": ["lead time", "supplier", "arrival", "purchase order", "procurement", "补货", "采购", "到货"],
"Campaign plan": ["promo", "campaign", "活动", "大促", "618", "11.11", "launch", "peak"],
"Seasonality / category": ["season", "seasonal", "类目", "category", "tag", "collection", "季节"],
}
RISK_KEYWORDS = {
"Stockout risk": ["stockout", "out of stock", "缺货", "断货", "lead time", "hot seller", "hero sku"],
"Aging / overstock risk": ["aged", "aging", "滞销", "overstock", "slow mover", "dead stock", "clearance"],
"Capital lock-up": ["cash", "capital", "holding cost", "inventory amount", "资金", "occupy cash"],
"Structural imbalance": ["mix", "structure", "结构", "long tail", "long-tail", "category balance", "assortment"],
"Promo readiness gap": ["promo", "campaign", "活动", "大促", "618", "11.11", "launch"],
}
PRIORITY_ORDERS = {
"Stockout prevention": ["Stockout risk", "Promo readiness gap", "Structural imbalance", "Aging / overstock risk"],
"Aging cleanup": ["Aging / overstock risk", "Capital lock-up", "Structural imbalance", "Stockout risk"],
"Promo readiness": ["Promo readiness gap", "Stockout risk", "Structural imbalance", "Aging / overstock risk"],
"Cash discipline": ["Capital lock-up", "Aging / overstock risk", "Structural imbalance", "Stockout risk"],
}
RISK_LIBRARY = {
"Stockout risk": {
"inspect": "Days of cover, top-SKU velocity, supplier lead time, and whether hero demand already outpaces inbound supply.",
"action": "Expedite replenishment, protect stock exposure, or shift demand into substitutes before the gap becomes customer-visible.",
},
"Aging / overstock risk": {
"inspect": "Slow sell-through, aging buckets, markdown history, and whether inventory is stacking inside low-demand variants.",
"action": "Freeze rebuys, build a clearance or bundle lane, and reduce low-conviction stock before it drags the next cycle.",
},
"Capital lock-up": {
"inspect": "Inventory amount tied in slow or low-margin SKUs versus how much cash is protected by hero and core lines.",
"action": "Pause discretionary purchasing and redirect working capital toward faster, cleaner inventory turns.",
},
"Structural imbalance": {
"inspect": "The balance between hero, core, seasonal, and long-tail SKUs, plus whether category mix reflects current demand.",
"action": "Rebalance future buys, simplify weak tails, and stop letting low-value assortment complexity crowd out priority stock.",
},
"Promo readiness gap": {
"inspect": "Promo SKU demand, inbound timing, warehouse readiness, and whether featured products have enough protected stock.",
"action": "Lock a promo stock plan, narrow the hero SKU list, and align campaign promises to credible available inventory.",
},
}
class InventoryHealthAuditor:
def __init__(self, user_input: Any):
self.raw = user_input
self.text = _normalize_inputs(user_input)
self.lower = self.text.lower()
self.objective = _detect_one(self.text, OBJECTIVE_RULES, "Stockout prevention")
self.signals = _detect_many(
self.text,
SIGNAL_RULES,
["Inventory on hand", "Sales velocity", "Lead time / replenishment"],
limit=5,
)
self.risks = self._priority_risks()
def _priority_risks(self) -> List[str]:
defaults = PRIORITY_ORDERS[self.objective]
return _detect_many(self.text, RISK_KEYWORDS, defaults, limit=4)
def _action_plan(self) -> List[str]:
plan_map = {
"Stockout prevention": [
"Week 1: isolate the hero SKUs with the thinnest days of cover and confirm inbound reality instead of planned inbound only.",
"Week 2: protect demand by limiting exposure, swapping substitutes, or prioritizing the cleanest replenishment path.",
"Weeks 3-4: rebalance future purchasing so the next cycle does not repeat the same shortage pattern.",
],
"Aging cleanup": [
"Week 1: rank aged and slow-moving SKUs by cash tied up, not just by unit count.",
"Week 2: launch the simplest recovery move first, such as bundle, markdown lane, or purchase freeze.",
"Weeks 3-4: remove repeat causes by tightening assortment and reducing weak replenishment habits.",
],
"Promo readiness": [
"Week 1: confirm the promo hero list, inbound timing, and which SKUs should be protected versus deprioritized.",
"Week 2: align campaign promises, warehouse capacity, and stock allocation before traffic ramps.",
"Weeks 3-4: monitor the featured assortment separately so promo demand does not hide broader imbalance.",
],
"Cash discipline": [
"Week 1: map where cash is trapped across slow, low-margin, or structurally weak SKUs.",
"Week 2: stop low-conviction purchases and prioritize actions that convert trapped stock into usable working capital.",
"Weeks 3-4: reset the buy logic so cash goes first to cleaner turns and higher-confidence inventory positions.",
],
}
return plan_map[self.objective]
def _assumptions(self) -> List[str]:
notes = [
"This report is heuristic and should be treated as an operator-facing decision brief, not a live planning system.",
"Final replenishment volume, purchase orders, and markdown decisions should remain human-approved.",
]
missing = [
signal
for signal in ["Sales velocity", "Lead time / replenishment", "Campaign plan"]
if signal not in self.signals
]
if missing:
notes.append(f"Signals that appear missing or only lightly referenced: {_join(missing)}.")
if "Seasonality / category" not in self.signals:
notes.append("If seasonality is material, re-check the output before treating slow movement as true structural weakness.")
return notes
def render(self) -> str:
meta = _load_skill_meta()
lines: List[str] = []
lines.append("# Inventory Health Audit Report")
lines.append("")
lines.append(f"**Skill description:** {meta['description']}")
lines.append(f"**Primary objective:** {self.objective}")
lines.append(f"**Signals referenced:** {_join(self.signals)}")
lines.append(f"**Priority lenses:** {_join(self.risks)}")
lines.append(f"**Input snapshot:** {self.text or 'No structured inventory notes were provided, so this report uses default weekly inventory-review assumptions.'}")
lines.append("")
lines.append("## Overview")
lines.append("- Treat inventory health as a balance among service level, cash discipline, and demand readiness, not as a pure stock-quantity question.")
lines.append(f"- Because the main objective is **{self.objective.lower()}**, the first action should target the smallest number of SKUs that create the largest operational or cash consequence.")
lines.append("- Separate hero, core, seasonal, and long-tail logic before making any broad purchase or clearance decision.")
lines.append("")
lines.append("## Priority Queue")
lines.append("| Priority | Risk type | What to inspect now | Recommended action |")
lines.append("|---|---|---|---|")
for idx, risk in enumerate(self.risks, 1):
details = RISK_LIBRARY[risk]
lines.append(f"| {idx} | {risk} | {details['inspect']} | {details['action']} |")
lines.append("")
lines.append("## 30-Day Action Plan")
for idx, step in enumerate(self._action_plan(), 1):
lines.append(f"{idx}. {step}")
lines.append("")
lines.append("## Assumptions and Limits")
for note in self._assumptions():
lines.append(f"- {note}")
return "\n".join(lines)
def handle(user_input: Any) -> str:
return InventoryHealthAuditor(user_input).render()
if __name__ == "__main__":
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import InventoryHealthAuditor, handle
def test_objective_detection_stockout_prevention():
auditor = InventoryHealthAuditor('Our hero SKU may stockout before 618 because lead time is 25 days and sales keep rising.')
assert auditor.objective == 'Stockout prevention'
def test_signal_detection_sales_and_lead_time():
auditor = InventoryHealthAuditor('Need an inventory review with sales velocity, supplier lead time, and current stock on hand.')
assert 'Sales velocity' in auditor.signals
assert 'Lead time / replenishment' in auditor.signals
assert 'Inventory on hand' in auditor.signals
def test_render_contains_required_sections():
output = handle('Check for aging inventory, overstock, and cash tied up in slow movers.')
assert output.startswith('# Inventory Health Audit Report')
assert '## Priority Queue' in output
assert '## 30-Day Action Plan' in output
def test_dict_input_supported():
output = handle({
'inventory': 'hero sku 120 units, tail sku 900 units',
'lead_time': '20 days',
'campaign': '618 promotion',
})
assert '# Inventory Health Audit Report' in output
assert '618 promotion' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Turn warehouse process notes, bottleneck observations, labor constraints, slotting issues, and service targets into a practical flow optimization brief with...
---
name: warehouse-flow-optimizer
description: Turn warehouse process notes, bottleneck observations, labor constraints, slotting issues, and service targets into a practical flow optimization brief with bottleneck hypotheses, quick wins, pilot ideas, and operating guardrails. Use when ecommerce ops teams, 3PL managers, or consultants need warehouse improvement guidance without live WMS, OMS, or labor-system integrations.
---
# Warehouse Flow Optimizer
## Overview
Use this skill to structure warehouse improvement work when the team has pain signals but not a full industrial-engineering study. It helps operators turn rough observations into a focused bottleneck and pilot plan.
This MVP is heuristic. It does **not** connect to a live WMS, OMS, labor system, or automation controller. It relies on the user's process notes, KPI summaries, and constraints.
## Trigger
Use this skill when the user wants to:
- reduce pick, pack, or dock bottlenecks
- improve slotting, replenishment, or travel-time efficiency
- stabilize shift throughput during peak periods
- diagnose why order cutoff performance or SLA adherence is slipping
- turn warehouse pain points into a practical improvement memo
### Example prompts
- "Help me identify the biggest warehouse bottleneck from these shift notes"
- "Create a quick-win plan for picking congestion and replenishment delays"
- "How should we think about slotting and labor balance before peak?"
- "Turn these warehouse KPI notes into an optimization brief"
## Workflow
1. Clarify the flow objective, operating window, and service target.
2. Normalize bottleneck signals such as queueing, travel time, pick accuracy, and space use.
3. Separate root-cause hypotheses across receiving, putaway, replenishment, picking, packing, and dock flow.
4. Recommend a short list of quick wins and one pilot path.
5. Return a markdown brief with assumptions, guardrails, and next steps.
## Inputs
The user can provide any mix of:
- throughput, pick rate, pack rate, or dock timing notes
- layout, slotting, replenishment, or congestion observations
- labor plan, shift coverage, or cross-training constraints
- inventory accuracy, stockout, or replenishment delay notes
- carrier cutoff, SLA, or peak-season timing pressure
- automation limits, capex limits, or fixed-layout constraints
## Outputs
Return a markdown warehouse brief with:
- primary bottleneck focus
- flow summary and bottleneck map
- root-cause hypotheses
- quick wins and pilot moves
- operating guardrails and monitoring cues
- assumptions, confidence notes, and limits
## Safety
- Do not claim access to live WMS or labor data.
- Do not present bottleneck hypotheses as proven without observation or measurement.
- Avoid recommending irreversible layout or automation changes from sparse notes alone.
- Keep staffing, safety, and capex decisions human-approved.
- Downgrade confidence when KPI definitions or zone-level detail are unclear.
## Best-fit Scenarios
- ecommerce warehouses or 3PL nodes looking for fast operational triage
- teams preparing for peak, SLA recovery, or labor rebalancing
- operators who need a simple improvement brief before deeper engineering work
- consultants framing a first-pass warehouse optimization plan
## Not Ideal For
- greenfield facility design or detailed simulation modeling
- robotics control logic or automation system tuning
- fully quantified industrial-engineering studies with time-motion data
- workflows that must write changes directly into WMS or labor tools
## Acceptance Criteria
- Return markdown text.
- Include bottleneck, action, pilot, and assumption sections.
- Keep the advisory framing explicit.
- Make the output practical for warehouse operators and ops leaders.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
ActionInput = Union[str, Dict[str, Any]]
FOCUS_RULES = {
'Receiving / Dock Flow': ['receiving', 'dock', 'unload', 'carrier', 'cutoff'],
'Slotting / Putaway': ['slotting', 'putaway', 'bin', 'layout', 'travel time'],
'Picking / Packing': ['picking', 'pick', 'packing', 'pack', 'wave', 'congestion'],
'Replenishment Control': ['replenishment', 'stockout', 'top-off', 'forward pick', 'refill'],
'Labor Balancing': ['labor', 'shift', 'headcount', 'cross-train', 'productivity'],
}
SIGNAL_RULES = {
'Queue / Congestion': ['queue', 'congestion', 'wave', 'jam', 'backlog'],
'Travel Time Waste': ['travel time', 'walking', 'distance', 'layout', 'slotting'],
'Pick / Pack Accuracy': ['accuracy', 'mispick', 'error', 'rework', 'wrong item'],
'Inventory Accuracy / Replenishment': ['inventory', 'stockout', 'replenishment', 'count', 'forward pick'],
'Space Utilization': ['space', 'cube', 'slot', 'staging', 'overflow'],
'Labor Imbalance': ['labor', 'shift', 'staffing', 'cross-train', 'headcount'],
'Cutoff / Service Risk': ['cutoff', 'sla', 'carrier', 'late', 'same day'],
}
CONSTRAINT_RULES = {
'Fixed layout': ['fixed layout', 'cannot move', 'layout constraint'],
'Headcount cap': ['headcount', 'labor cap', 'hiring freeze'],
'Peak pressure': ['peak', 'holiday', 'surge'],
'Service-level commitment': ['sla', 'cutoff', 'same day', 'next day'],
'Low automation flexibility': ['manual', 'no automation', 'limited automation'],
}
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for keyword in keywords if keyword in text) for name, keywords in rules.items()}
def _join_list(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class WarehouseFlowOptimizer:
def __init__(self, user_input: ActionInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.focus = self._detect_focus()
self.signals = self._detect_signals()
self.constraints = self._detect_constraints()
def _normalize_input(self, user_input: ActionInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['objective', 'zones', 'signals', 'constraints', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_focus(self) -> str:
scores = _score_rules(self.lower, FOCUS_RULES)
best = max(scores, key=scores.get)
return best if scores[best] > 0 else 'Picking / Packing'
def _detect_signals(self) -> List[str]:
matched = [name for name, keywords in SIGNAL_RULES.items() if any(keyword in self.lower for keyword in keywords)]
return matched or ['Queue / Congestion', 'Travel Time Waste', 'Labor Imbalance']
def _detect_constraints(self) -> List[str]:
matched = [name for name, keywords in CONSTRAINT_RULES.items() if any(keyword in self.lower for keyword in keywords)]
base = matched or ['Service-level commitment', 'Headcount cap']
if 'Service-level commitment' not in base:
base.append('Service-level commitment')
return base[:4]
def _root_causes(self) -> List[str]:
causes = [
'The visible bottleneck may be downstream of poor replenishment timing, weak slotting, or handoff friction rather than raw labor effort alone.',
'Throughput pain often compounds when the same people absorb queue clearing, exception handling, and normal flow work at the same time.',
'If staging and forward-pick locations are unstable, the team pays twice through search time and rework.',
]
if self.focus == 'Receiving / Dock Flow':
causes.append('Dock congestion usually reflects arrival clustering, unload sequencing, or delayed putaway capacity, not just carrier volume.')
elif self.focus == 'Slotting / Putaway':
causes.append('Travel waste often points to slotting drift where demand velocity changed faster than location logic.')
elif self.focus == 'Replenishment Control':
causes.append('Forward-pick starvation can make otherwise healthy picking teams look slow and inconsistent.')
return causes[:4]
def _quick_wins(self) -> List[str]:
return [
'Stabilize one high-volume zone first instead of redesigning the entire building at once.',
'Create a simple exception lane so abnormal orders stop stealing capacity from normal flow.',
'Tighten the replenishment trigger for the fastest movers before the next heavy shift window.',
'Use one visual control for queue status, aged backlog, and near-cutoff risk so supervisors react earlier.',
]
def _pilot_plan(self) -> List[str]:
return [
'Week 1: baseline one zone or process with a small set of comparable shift metrics.',
'Week 2: trial one slotting, replenishment, or labor-balance change without adding multiple new variables.',
'Week 3: compare throughput, errors, backlog age, and cutoff hit rate against the baseline.',
'Week 4: expand only if the pilot improved both speed and operational stability, not just short bursts of output.',
]
def _assumptions(self) -> List[str]:
notes = [
'This brief is heuristic and depends on the quality of the user-supplied KPI and process notes.',
'No live WMS, OMS, labor, or automation telemetry was accessed.',
'Bottlenecks are inferred from described symptoms and should be confirmed with floor observation or measurement.',
]
if 'Inventory Accuracy / Replenishment' not in self.signals:
notes.append('Replenishment detail appears limited, so some root-cause attribution may be incomplete.')
return notes
def render(self) -> str:
lines: List[str] = []
lines.append('# Warehouse Flow Optimizer Brief')
lines.append('')
lines.append(f'**Primary focus:** {self.focus}')
lines.append(f'**Flow signals referenced:** {_join_list(self.signals)}')
lines.append(f'**Operating constraints:** {_join_list(self.constraints)}')
lines.append('**Method note:** This is a heuristic warehouse optimization brief. No live WMS, labor, or automation data was accessed.')
lines.append('')
lines.append('## Flow Summary')
lines.append('- Start with the narrowest bottleneck that most strongly affects throughput or service risk.')
lines.append('- Separate structural flow waste from temporary peak pressure or one-off disruptions.')
lines.append('- Prefer local pilots that are measurable and reversible before scaling changes across the building.')
lines.append('')
lines.append('## Root-Cause Hypotheses')
for bullet in self._root_causes():
lines.append(f'- {bullet}')
lines.append('')
lines.append('## Quick Wins')
for idx, item in enumerate(self._quick_wins(), 1):
lines.append(f'{idx}. {item}')
lines.append('')
lines.append('## 30-Day Pilot Plan')
for idx, item in enumerate(self._pilot_plan(), 1):
lines.append(f'{idx}. {item}')
lines.append('')
lines.append('## Guardrails and Monitoring')
for constraint in self.constraints:
lines.append(f'- {constraint}: check that any improvement move respects this operational boundary.')
lines.append('')
lines.append('## Assumptions and Limits')
for note in self._assumptions():
lines.append(f'- {note}')
return '\n'.join(lines)
def handle(user_input: ActionInput) -> str:
return WarehouseFlowOptimizer(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import WarehouseFlowOptimizer, handle
def test_focus_detection_replenishment():
optimizer = WarehouseFlowOptimizer('Our forward-pick stockouts and replenishment delays are slowing every shift.')
assert optimizer.focus == 'Replenishment Control'
def test_signal_detection_labor_and_cutoff():
optimizer = WarehouseFlowOptimizer('We have headcount imbalance, late carrier cutoff misses, and staging congestion.')
assert 'Labor Imbalance' in optimizer.signals
assert 'Cutoff / Service Risk' in optimizer.signals
def test_dict_input_supported():
output = handle({
'objective': 'reduce picking congestion',
'zones': ['pick module', 'packing wall'],
'signals': ['queue', 'travel time'],
'notes': 'Need quick wins before peak season.',
})
assert '# Warehouse Flow Optimizer Brief' in output
assert '## 30-Day Pilot Plan' in output
def test_render_contains_sections():
output = handle('Help me improve slotting, walking distance, and queueing before next day SLA starts slipping.')
assert output.startswith('# Warehouse Flow Optimizer Brief')
assert '## Quick Wins' in output
assert '## Assumptions and Limits' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Turn pricing notes, margin pressure, competitor context, discount behavior, and channel constraints into a practical pricing strategy brief with objective fr...
---
name: pricing-strategy-advisor
description: Turn pricing notes, margin pressure, competitor context, discount behavior, and channel constraints into a practical pricing strategy brief with objective framing, move options, test ideas, and guardrails. Use when ecommerce teams, brand operators, category managers, or consultants need pricing guidance without live competitor crawlers, ERP feeds, or analytics APIs.
---
# Pricing Strategy Advisor
## Overview
Use this skill to structure pricing decisions when the team has partial information but still needs a clear commercial recommendation. It helps with pricing architecture, margin recovery, launch pricing, and discount discipline.
This MVP is heuristic. It does **not** fetch live competitor prices, cost feeds, marketplace data, or conversion dashboards. It relies on the user's supplied context and assumptions.
## Trigger
Use this skill when the user wants to:
- decide whether to raise, hold, narrow, or tier prices
- recover margin without blindly killing conversion
- tighten discount discipline or promo hygiene
- design a price ladder, bundle, or good-better-best structure
- prepare a pricing memo before a launch, seasonal reset, or negotiation
### Example prompts
- "Help me decide whether we should raise prices or cut discount depth"
- "Design a better pricing ladder for our core line"
- "We have competitor pressure and rising costs. What pricing moves make sense?"
- "Create a pricing test plan for this product launch"
## Workflow
1. Clarify the pricing objective and what business trade-off matters most.
2. Normalize the strongest signals, such as cost, conversion, competitor pressure, and discount behavior.
3. Separate structural pricing issues from temporary promo, inventory, or channel noise.
4. Recommend a short list of moves with guardrails and experiment ideas.
5. Return a markdown strategy brief with risks, metrics, and assumptions.
## Inputs
The user can provide any mix of:
- current and target prices, margins, or cost changes
- competitor references or marketplace context
- conversion, volume, AOV, or bundle behavior
- promo cadence, coupon usage, or markdown history
- channel constraints such as MAP, retailer parity, or marketplace conflict
- inventory pressure, premium positioning, or launch goals
## Outputs
Return a markdown pricing brief with:
- primary pricing objective
- pricing posture summary
- recommended moves and trade-offs
- test design ideas and guardrails
- metrics to monitor and escalation notes
- assumptions, confidence notes, and limits
## Safety
- Do not claim access to live competitor or conversion data.
- Do not present price elasticity as proven unless the user supplies evidence.
- Avoid auto-approving permanent price increases, channel exceptions, or MAP violations.
- Keep final pricing changes human-approved.
- Downgrade confidence when channel conflict or unit economics are unclear.
## Best-fit Scenarios
- DTC and marketplace brands reviewing price strategy quarterly or around major campaigns
- operators who need a structured recommendation before changing discount or ladder logic
- founders balancing growth, margin, and positioning without a full pricing stack
- consultants preparing a first-pass pricing memo
## Not Ideal For
- real-time competitor repricing systems
- heavily regulated pricing environments that require legal review
- highly quantitative elasticity modeling with experimental control data
- workflows that must write prices directly into commerce systems
## Acceptance Criteria
- Return markdown text.
- Include objective, recommendation, guardrail, and monitoring sections.
- Keep the advisory framing explicit.
- Make trade-offs practical for operators and decision-makers.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
ActionInput = Union[str, Dict[str, Any]]
OBJECTIVE_RULES = {
'Launch Pricing': ['launch', 'new product', 'intro', 'introduction', 'gtm'],
'Discount Discipline': ['discount', 'coupon', 'promo', 'markdown', 'sale'],
'Price Architecture': ['price ladder', 'architecture', 'good better best', 'tier', 'premium', 'entry price'],
'Margin Recovery': ['margin', 'profit', 'cost', 'gross margin', 'contribution'],
'Growth Push': ['growth', 'volume', 'share', 'conversion', 'acquisition'],
}
SIGNAL_RULES = {
'Cost / Margin Pressure': ['cost', 'margin', 'profit', 'contribution'],
'Elasticity / Conversion': ['conversion', 'cvr', 'elasticity', 'volume', 'sell-through', 'sell through'],
'Competitor Pressure': ['competitor', 'market', 'benchmark', 'price war', 'marketplace'],
'Channel Conflict': ['channel', 'retailer', 'amazon', 'marketplace', 'map', 'parity'],
'Promo Fatigue': ['discount', 'coupon', 'promo', 'markdown', 'sale fatigue'],
'Bundle / Pack Mix': ['bundle', 'pack', 'multi-buy', 'pack size', 'tier'],
}
GUARDRAIL_RULES = {
'Minimum margin floor': ['margin floor', 'gross margin', 'contribution'],
'Brand or MAP protection': ['map', 'brand', 'premium', 'positioning'],
'Channel parity discipline': ['parity', 'retailer', 'marketplace', 'channel'],
'Inventory aging pressure': ['aged', 'aging', 'clearance', 'inventory'],
'Launch learning speed': ['launch', 'test', 'learn', 'pilot'],
}
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for keyword in keywords if keyword in text) for name, keywords in rules.items()}
def _join_list(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class PricingStrategyAdvisor:
def __init__(self, user_input: ActionInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.objective = self._detect_objective()
self.signals = self._detect_signals()
self.guardrails = self._detect_guardrails()
def _normalize_input(self, user_input: ActionInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['objective', 'prices', 'signals', 'constraints', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_objective(self) -> str:
scores = _score_rules(self.lower, OBJECTIVE_RULES)
best = max(scores, key=scores.get)
return best if scores[best] > 0 else 'Price Architecture'
def _detect_signals(self) -> List[str]:
matched = [name for name, keywords in SIGNAL_RULES.items() if any(keyword in self.lower for keyword in keywords)]
return matched or ['Cost / Margin Pressure', 'Elasticity / Conversion', 'Competitor Pressure']
def _detect_guardrails(self) -> List[str]:
matched = [name for name, keywords in GUARDRAIL_RULES.items() if any(keyword in self.lower for keyword in keywords)]
base = matched or ['Minimum margin floor', 'Brand or MAP protection']
if 'Minimum margin floor' not in base:
base.append('Minimum margin floor')
return base[:4]
def _moves(self) -> List[str]:
if self.objective == 'Launch Pricing':
return [
'Start with a learnable price band rather than the deepest possible discount at launch.',
'Use one clean introductory offer and a short review point instead of stacking coupons early.',
'Protect the premium narrative if the product depends on perceived quality, not just trial volume.',
'Predefine what evidence would justify a price move after the first demand cycle.',
]
if self.objective == 'Discount Discipline':
return [
'Reduce promo sprawl by tightening where and when discounts appear, rather than cutting list price first.',
'Separate traffic-driving offers from margin-destructive blanket markdowns.',
'Reserve the deepest discounting for genuinely aged or tactical inventory, not habit-driven cadence.',
'Use bundles, packs, or threshold offers to preserve unit economics before deeper discounting.',
]
if self.objective == 'Margin Recovery':
return [
'Recover margin first through architecture, pack, mix, or selective price moves before blunt across-the-board increases.',
'Raise prices where differentiation is strongest and conversion is less discount-dependent.',
'Keep an entry-point option if it protects acquisition while premium or bundle offers recover margin.',
'Explain the reason for any price change in plain business terms so the team can defend it internally and externally.',
]
if self.objective == 'Growth Push':
return [
'Use pricing to sharpen the conversion path, not to subsidize low-quality demand indefinitely.',
'Anchor volume on the hero offers that create repeatable acquisition or retention value.',
'Use bundles or tier progression to lift AOV before defaulting to simpler discount depth.',
'Keep the move reversible so learning arrives before margin damage compounds.',
]
return [
'Clarify the good-better-best logic so each tier earns its place through clear value separation.',
'Reduce overlap between nearby price points that confuse buyers or dilute premium progression.',
'Use pack, bundle, or service differences to support tiering instead of cosmetic SKU inflation.',
'Make sure the entry offer acquires, the core offer monetizes, and the premium offer signals upside.',
]
def _test_plan(self) -> List[str]:
return [
'Run one focused pricing test on a clearly scoped SKU family or offer set, not across the full catalog at once.',
'Define the primary win metric before launch, such as contribution margin, conversion, AOV, or net revenue per session.',
'Track new versus returning customer behavior separately so short-term lift is not mistaken for structural improvement.',
'Set a review checkpoint early enough to reverse the move if channel conflict or margin damage appears quickly.',
]
def _monitoring(self) -> List[str]:
items = [
'Monitor gross margin, unit contribution, conversion rate, and AOV together instead of debating only one metric.',
'Watch refund, cancel, and service signal changes after pricing moves so hidden demand quality problems surface early.',
'Review channel-level divergence, especially if marketplaces or retail partners react differently than DTC.',
]
if 'Competitor Pressure' in self.signals:
items.append('Treat competitor references as context, not automatic permission to copy a price war.')
return items[:4]
def _assumptions(self) -> List[str]:
notes = [
'This brief is heuristic and depends on the user-supplied view of costs, pricing, and channel context.',
'No live competitor crawling, ERP costing, or analytics feed was accessed.',
'Elasticity is inferred from context unless the user provides direct experimental evidence.',
]
if 'Channel Conflict' not in self.signals:
notes.append('Channel conflict appears lightly referenced, so parity and retailer reaction risk may be understated.')
return notes
def render(self) -> str:
lines: List[str] = []
lines.append('# Pricing Strategy Advisor Brief')
lines.append('')
lines.append(f'**Primary objective:** {self.objective}')
lines.append(f'**Pricing signals referenced:** {_join_list(self.signals)}')
lines.append(f'**Guardrails:** {_join_list(self.guardrails)}')
lines.append('**Method note:** This is a heuristic pricing strategy brief. No live competitor, ERP, or analytics data was accessed.')
lines.append('')
lines.append('## Pricing Posture Summary')
lines.append('- Start with the business trade-off that matters most: margin recovery, growth, positioning, or cleaner architecture.')
lines.append('- Separate structural price problems from temporary inventory, promo, or channel noise.')
lines.append('- Prefer moves that can be tested and reversed before they become permanent habits.')
lines.append('')
lines.append('## Recommended Moves')
for idx, move in enumerate(self._moves(), 1):
lines.append(f'{idx}. {move}')
lines.append('')
lines.append('## Test Design')
for bullet in self._test_plan():
lines.append(f'- {bullet}')
lines.append('')
lines.append('## Guardrails and Escalation')
for guardrail in self.guardrails:
lines.append(f'- {guardrail}: confirm ownership and escalation before any durable price change.')
lines.append('')
lines.append('## Metrics to Monitor')
for item in self._monitoring():
lines.append(f'- {item}')
lines.append('')
lines.append('## Assumptions and Limits')
for note in self._assumptions():
lines.append(f'- {note}')
return '\n'.join(lines)
def handle(user_input: ActionInput) -> str:
return PricingStrategyAdvisor(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import PricingStrategyAdvisor, handle
def test_objective_detection_discount():
advisor = PricingStrategyAdvisor('Our coupon cadence is out of control and markdowns are hurting margin.')
assert advisor.objective == 'Discount Discipline'
def test_signal_detection_competitor_and_channel():
advisor = PricingStrategyAdvisor('We face competitor pressure on Amazon and need better marketplace parity.')
assert 'Competitor Pressure' in advisor.signals
assert 'Channel Conflict' in advisor.signals
def test_dict_input_supported():
output = handle({
'objective': 'margin recovery',
'prices': ['current price 129', 'target margin 62%'],
'signals': ['cost increase', 'conversion sensitivity'],
'notes': 'Need a recommendation before next quarter planning.',
})
assert '# Pricing Strategy Advisor Brief' in output
assert '## Recommended Moves' in output
def test_render_contains_sections():
output = handle('Design a better price ladder and bundle logic for our core product line.')
assert output.startswith('# Pricing Strategy Advisor Brief')
assert '## Test Design' in output
assert '## Assumptions and Limits' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Turn sales notes, trend signals, seasonal context, promo plans, and inventory constraints into a practical demand forecast brief with base, upside, and downs...
---
name: trend-demand-forecaster
description: Turn sales notes, trend signals, seasonal context, promo plans, and inventory constraints into a practical demand forecast brief with base, upside, and downside scenarios, leading indicators, and replenishment cues. Use when planners, ecommerce operators, founders, or consultants need forecasting support without live ERP, BI, ads, or marketplace APIs.
---
# Trend Demand Forecaster
## Overview
Use this skill to convert rough demand signals into a practical forecast narrative. It is built for teams that need a fast planning layer for the next few weeks, quarter, or seasonal window.
This MVP is heuristic. It does **not** pull live sales, ads, weather, ERP, marketplace, or competitor data. It relies on the user's provided notes, exports, and planning context.
## Trigger
Use this skill when the user wants to:
- forecast demand for the next month, quarter, or seasonal event
- estimate promo lift or post-promo normalization
- plan replenishment against demand uncertainty
- interpret whether demand is recovering, stabilizing, or softening
- turn messy trend notes into a base, upside, downside planning brief
### Example prompts
- "Forecast next month's demand using these sales and inventory notes"
- "Build a base, upside, downside demand view for our holiday campaign"
- "Should we buy deeper inventory or stay cautious?"
- "Help me interpret whether demand is rebounding or just promo noise"
## Workflow
1. Clarify the planning question, decision horizon, and risk tolerance.
2. Normalize the strongest signals, such as traffic, orders, conversion, price, inventory, and seasonality.
3. Separate baseline demand from promo distortion, stockout distortion, or one-off events.
4. Build base, upside, and downside scenarios with trigger conditions.
5. Return a markdown brief with indicators, action cues, and assumptions.
## Inputs
The user can provide any mix of:
- weekly or monthly sales summaries
- traffic, conversion, pricing, and promo notes
- stockout periods, inventory cover, or inbound timing
- launch, seasonal, holiday, or campaign context
- return rate, customer service, or marketplace feedback
- constraints such as cash, lead time, MOQ, or warehouse limits
## Outputs
Return a markdown forecast brief with:
- demand narrative and likely mode
- planning horizon and key signals
- base, upside, downside scenarios
- leading indicators to monitor
- inventory and commercial implications
- risk watchlist and next-step actions
- assumptions, confidence notes, and limits
## Safety
- Do not claim access to live systems or external trend feeds.
- Treat all scenarios as planning heuristics, not guaranteed forecasts.
- Do not auto-commit buys, budgets, or inventory transfers.
- Downgrade confidence when the user only provides promo-distorted or stockout-distorted history.
- Keep final purchasing and budget decisions human-approved.
## Best-fit Scenarios
- ecommerce teams planning 2 to 16 weeks ahead
- operators working from rough exports instead of a forecasting platform
- founders who need a quick demand-planning memo before placing inventory bets
- consultants preparing scenario-based planning recommendations
## Not Ideal For
- formal statistical forecasting that requires model calibration and backtesting
- highly granular store-SKU-day forecasting at enterprise scale
- workflows that require automatic PO creation or system sync
- regulated forecasts that need audited financial controls
## Acceptance Criteria
- Return markdown text.
- Include scenario, indicator, implication, and risk sections.
- Make the advisory and heuristic framing explicit.
- Keep the output practical for planning and replenishment decisions.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
ActionInput = Union[str, Dict[str, Any]]
MODE_RULES = {
'Promo Lift Planning': ['promo', 'promotion', 'campaign', 'holiday', 'launch', 'event', 'sale'],
'Replenishment Planning': ['inventory', 'replenish', 'stock', 'buy', 'inbound', 'purchase order', 'coverage'],
'Recovery / Slowdown Read': ['recover', 'recovery', 'rebound', 'drop', 'decline', 'slowdown', 'softness'],
'Baseline Forecast': ['forecast', 'baseline', 'plan', 'next month', 'next quarter', 'demand'],
}
SIGNAL_RULES = {
'Traffic / Order Volume': ['traffic', 'sessions', 'orders', 'volume', 'demand', 'gmv', 'sales'],
'Conversion / Funnel': ['conversion', 'cvr', 'checkout', 'cart', 'funnel'],
'Price / Promotion': ['price', 'pricing', 'discount', 'promo', 'coupon', 'markdown'],
'Inventory / Availability': ['inventory', 'stock', 'stockout', 'oos', 'coverage', 'inbound'],
'Returns / Service': ['return', 'refund', 'rating', 'review', 'service', 'complaint'],
'External Trend Context': ['seasonal', 'holiday', 'trend', 'weather', 'competitor', 'market'],
}
RISK_RULES = {
'Promo distortion': ['promo', 'discount', 'campaign', 'sale'],
'Stockout distortion': ['stockout', 'oos', 'out of stock', 'inventory'],
'Seasonality shift': ['seasonal', 'holiday', 'weather'],
'Assortment change': ['launch', 'new sku', 'new product', 'bundle', 'assortment'],
'Weak data quality': ['rough notes', 'estimate', 'limited data', 'partial'],
}
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for keyword in keywords if keyword in text) for name, keywords in rules.items()}
def _join_list(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class TrendDemandForecaster:
def __init__(self, user_input: ActionInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.mode = self._detect_mode()
self.horizon = self._detect_horizon()
self.signals = self._detect_signals()
self.risks = self._detect_risks()
def _normalize_input(self, user_input: ActionInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['objective', 'horizon', 'signals', 'constraints', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_mode(self) -> str:
scores = _score_rules(self.lower, MODE_RULES)
best = max(scores, key=scores.get)
return best if scores[best] > 0 else 'Baseline Forecast'
def _detect_horizon(self) -> str:
if any(keyword in self.lower for keyword in ['quarter', 'q1', 'q2', 'q3', 'q4']):
return 'Next quarter'
if any(keyword in self.lower for keyword in ['holiday', 'seasonal', 'summer', 'winter', 'festival']):
return 'Seasonal window'
if any(keyword in self.lower for keyword in ['week', '4 week', '8 week', 'month']):
return 'Next 4 to 8 weeks'
return 'Next 4 to 8 weeks'
def _detect_signals(self) -> List[str]:
matched = [name for name, keywords in SIGNAL_RULES.items() if any(keyword in self.lower for keyword in keywords)]
return matched or ['Traffic / Order Volume', 'Conversion / Funnel', 'Inventory / Availability']
def _detect_risks(self) -> List[str]:
matched = [name for name, keywords in RISK_RULES.items() if any(keyword in self.lower for keyword in keywords)]
if not matched:
matched = ['Weak data quality']
if 'Weak data quality' not in matched:
matched.append('Weak data quality')
return matched[:4]
def _scenario_rows(self) -> List[Dict[str, str]]:
if self.mode == 'Promo Lift Planning':
return [
{'scenario': 'Base', 'implication': 'Promo creates a temporary lift, then demand normalizes toward the recent underlying run rate.', 'cue': 'Buy enough to protect hero SKUs, but avoid assuming the full event lift persists.'},
{'scenario': 'Upside', 'implication': 'Campaign resonance and clean inventory availability sustain higher-than-expected sell-through.', 'cue': 'Prepare a fast reorder or transfer path for top products instead of broad overbuying.'},
{'scenario': 'Downside', 'implication': 'Lift is shallow or highly discount-dependent, so demand falls back quickly after the event.', 'cue': 'Keep markdown exposure and tail inventory tightly controlled.'},
]
if self.mode == 'Replenishment Planning':
return [
{'scenario': 'Base', 'implication': 'Demand stays close to the recent steady-state trend and the current inbound plan mostly fits.', 'cue': 'Replenish core SKUs first and defer speculative tail buys.'},
{'scenario': 'Upside', 'implication': 'Availability improves and conversion holds, creating a cleaner sell-through path than recent history suggests.', 'cue': 'Secure supplier flexibility for the top-demand families.'},
{'scenario': 'Downside', 'implication': 'Demand softens or lead-time risk rises, leaving slower inventory exposed.', 'cue': 'Reduce open-to-buy on low-confidence or low-margin items.'},
]
if self.mode == 'Recovery / Slowdown Read':
return [
{'scenario': 'Base', 'implication': 'Demand is stabilizing, but the recovery is still fragile and uneven by channel or SKU.', 'cue': 'Monitor leading indicators for two more cycles before making aggressive commitments.'},
{'scenario': 'Upside', 'implication': 'Improving traffic, conversion, and service quality point to a real recovery rather than noise.', 'cue': 'Lean into the strongest channels and products with controlled inventory expansion.'},
{'scenario': 'Downside', 'implication': 'The apparent rebound is mostly explained by promo timing, stock normalization, or one-off factors.', 'cue': 'Preserve cash and resist extrapolating a short-term bounce.'},
]
return [
{'scenario': 'Base', 'implication': 'Demand tracks the recent underlying trend after adjusting for major one-off distortions.', 'cue': 'Plan around the dependable core rather than headline spikes.'},
{'scenario': 'Upside', 'implication': 'Traffic quality, conversion, or mix improves enough to support a stronger demand curve.', 'cue': 'Use reorder triggers, not blanket optimism, to capture the upside.'},
{'scenario': 'Downside', 'implication': 'Soft traffic, weaker conversion, or operational friction drags demand below the current baseline.', 'cue': 'Protect margin and cash by slowing low-conviction buys.'},
]
def _leading_indicators(self) -> List[str]:
indicators = {
'Traffic / Order Volume': 'Track weekly sessions, order count, and top-SKU daily demand pace against a recent baseline.',
'Conversion / Funnel': 'Watch conversion rate, add-to-cart quality, and checkout completion for sign changes before revenue fully moves.',
'Price / Promotion': 'Separate true baseline demand from discount-driven lift and measure normalization after the offer ends.',
'Inventory / Availability': 'Check stockout hours, in-stock rate, and inbound timing so availability issues are not mistaken for weak demand.',
'Returns / Service': 'Use return reasons, complaint spikes, and rating changes as early warning signals for demand quality erosion.',
'External Trend Context': 'Confirm whether seasonality, holiday timing, or market trend changes are amplifying or suppressing demand.',
}
return [indicators[name] for name in self.signals][:4]
def _implications(self) -> List[str]:
actions = [
'Protect the core demand signal first, then layer scenario-specific inventory or budget actions on top.',
'Keep hero SKU availability cleaner than the long tail, because forecasting error matters most where demand concentrates.',
'Use scenario triggers to decide when to accelerate, pause, or defer incremental buys.',
]
if self.mode == 'Promo Lift Planning':
actions.append('Treat event demand as a pulse unless repeat-order quality or post-event conversion proves otherwise.')
elif self.mode == 'Replenishment Planning':
actions.append('Bias open-to-buy toward fast, flexible replenishment rather than broad speculative depth.')
else:
actions.append('Document what would count as evidence of a true recovery, not just a good week.')
return actions
def _assumptions(self) -> List[str]:
notes = [
'This brief is heuristic and depends on the quality of the user-supplied history and planning notes.',
'No live system, marketplace, ad platform, or external trend source was accessed.',
'Scenario ranges are decision aids, not statistical confidence intervals.',
]
if 'Inventory / Availability' not in self.signals:
notes.append('Availability details were limited, so lost-demand and true-baseline separation may be weak.')
return notes
def render(self) -> str:
lines: List[str] = []
lines.append('# Trend Demand Forecast Brief')
lines.append('')
lines.append(f'**Primary mode:** {self.mode}')
lines.append(f'**Planning horizon:** {self.horizon}')
lines.append(f'**Signals referenced:** {_join_list(self.signals)}')
lines.append(f'**Main risks:** {_join_list(self.risks)}')
lines.append('**Method note:** This is a heuristic demand-planning brief. No live ERP, BI, ads, or marketplace data was accessed.')
lines.append('')
lines.append('## Demand Narrative')
lines.append('- Start from the likely underlying trend, then adjust for promo noise, stock distortions, and seasonality.')
lines.append('- Treat the forecast as a planning range, not a single precise number.')
lines.append('- Focus on what would change the next buy, budget, or availability decision.')
lines.append('')
lines.append('## Scenario View')
lines.append('| Scenario | What it implies | Action cue |')
lines.append('|---|---|---|')
for row in self._scenario_rows():
lines.append(f'| {row["scenario"]} | {row["implication"]} | {row["cue"]} |')
lines.append('')
lines.append('## Leading Indicators')
for idx, item in enumerate(self._leading_indicators(), 1):
lines.append(f'{idx}. {item}')
lines.append('')
lines.append('## Inventory and Commercial Implications')
for bullet in self._implications():
lines.append(f'- {bullet}')
lines.append('')
lines.append('## Risk Watchlist')
for risk in self.risks:
lines.append(f'- {risk}: validate whether this is distorting the recent signal before scaling any commitment.')
lines.append('')
lines.append('## Assumptions and Limits')
for note in self._assumptions():
lines.append(f'- {note}')
return '\n'.join(lines)
def handle(user_input: ActionInput) -> str:
return TrendDemandForecaster(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import TrendDemandForecaster, handle
def test_mode_detection_promo():
forecaster = TrendDemandForecaster('Build a holiday promo demand forecast for our campaign launch and sale period.')
assert forecaster.mode == 'Promo Lift Planning'
def test_signal_detection_inventory_and_conversion():
forecaster = TrendDemandForecaster('Review conversion, checkout rate, stock coverage, and inbound risk for next month.')
assert 'Conversion / Funnel' in forecaster.signals
assert 'Inventory / Availability' in forecaster.signals
def test_dict_input_supported():
output = handle({
'objective': 'replenishment forecast',
'horizon': 'next quarter',
'signals': ['inventory', 'traffic'],
'notes': 'Need a cautious base, upside, downside view before placing purchase orders.',
})
assert '# Trend Demand Forecast Brief' in output
assert '## Scenario View' in output
def test_render_contains_sections():
output = handle('Forecast whether demand is recovering or just normalizing after stockouts and markdowns.')
assert output.startswith('# Trend Demand Forecast Brief')
assert '## Leading Indicators' in output
assert '## Assumptions and Limits' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Analyze product return or refund notes and turn them into a structured reason taxonomy, root-cause view, and fix-priority brief. Use when a team needs to und...
---
name: return-reason-miner
description: Analyze product return or refund notes and turn them into a structured reason taxonomy, root-cause view, and fix-priority brief. Use when a team needs to understand return drivers, separate product versus operations issues, or plan cross-functional remediation without live OMS, WMS, ERP, or BI access.
---
# Return Reason Miner
## Overview
Use this skill to convert messy return notes into a reason-based operating brief. It helps cluster return causes, identify likely root causes, suggest fix priorities, and frame what product, operations, CX, or merchandising teams should review next.
This MVP is heuristic. It does **not** connect to live order systems, warehouse systems, review platforms, or return portals. It relies on the user's provided return notes, product context, and issue patterns.
## Trigger
Use this skill when the user wants to:
- classify why products are being returned or refunded
- separate quality, fit, expectation, shipping, and fulfillment-related return drivers
- create a weekly or launch-period return review brief
- find likely cross-functional fixes instead of only counting return volume
- summarize rough return notes into a clear action plan
### Example prompts
- "Help me analyze our top return reasons for apparel"
- "Turn these refund notes into a root-cause brief"
- "What should product and ops teams look at if late delivery and wrong items are rising?"
- "Create a weekly return reason summary for our beauty line"
## Workflow
1. Capture the review mode, product context, and return evidence.
2. Choose the likely reason clusters and root-cause hypotheses.
3. Separate product issues from merchandising, fulfillment, and service issues.
4. Prioritize fixes by recurrence, severity, and controllability.
5. Return a markdown return-reason brief with cross-functional actions.
## Inputs
The user can provide any mix of:
- return notes, refund summaries, or customer feedback excerpts
- product context such as apparel, beauty, electronics, home goods, or food
- order, fulfillment, packaging, quality, sizing, and expectation notes
- launch period, promo period, or seasonal context
- any known operational changes such as new vendor, warehouse shift, or policy change
## Outputs
Return a markdown brief with:
- return pattern summary
- reason taxonomy table
- root-cause hypotheses
- fix priorities
- cross-functional action plan
- assumptions and limits
## Safety
- Do not claim access to live order, warehouse, or return systems.
- Treat root causes as hypotheses unless the evidence is strong and repeated.
- Do not fabricate percentages, defect rates, or financial impact.
- Final product, vendor, packaging, and policy decisions remain human-approved.
## Best-fit Scenarios
- ecommerce teams running weekly return or refund reviews
- operators trying to cut preventable returns without overreacting to noise
- product and CX teams needing a shared root-cause frame
## Not Ideal For
- automated returns processing or refund execution
- regulated product quality investigations that need formal QA evidence
- precise financial modeling of returns without structured data
## Acceptance Criteria
- Return markdown text.
- Include taxonomy, root-cause, action, and limits sections.
- Keep the heuristic framing explicit.
- Make the output practical for product, ops, and CX owners.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Sequence, Union
ReturnInput = Union[str, Dict[str, Any]]
REVIEW_MODE_RULES = {
'Weekly Return Review': ['weekly', 'week', 'recurring'],
'Launch Issue Triage': ['launch', 'new product', 'debut'],
'Seasonal Trend Scan': ['seasonal', 'holiday', 'promo period', 'campaign'],
'Vendor Quality Escalation': ['vendor', 'supplier', 'factory', 'quality issue'],
}
PRODUCT_CONTEXT_RULES = {
'Apparel': ['apparel', 'size', 'fit', 'clothing'],
'Beauty / Personal Care': ['beauty', 'skincare', 'cosmetic'],
'Electronics': ['electronic', 'device', 'charger', 'battery'],
'Home goods': ['home', 'kitchen', 'furniture', 'decor'],
'Food / Perishable': ['food', 'perishable', 'fresh', 'frozen'],
}
REASON_RULES = {
'Quality defect': ['defect', 'broken', 'faulty', 'quality issue'],
'Size or fit mismatch': ['size', 'fit', 'too small', 'too big'],
'Expectation mismatch': ['not as described', 'different than expected', 'color mismatch', 'expectation'],
'Shipping damage': ['damaged', 'crushed', 'leaked', 'shipping damage'],
'Wrong item or fulfillment error': ['wrong item', 'wrong color', 'wrong size', 'missing item', 'fulfillment'],
'Late delivery or timing miss': ['late', 'delay', 'missed', 'too slow'],
}
SIGNAL_RULES = {
'Return notes and free text': ['return note', 'refund note', 'feedback', 'reason'],
'SKU or variant clues': ['sku', 'variant', 'size', 'color'],
'Packaging or warehouse clues': ['warehouse', 'packaging', 'shipment', 'box'],
'Review or rating clues': ['review', 'rating', 'star'],
'Season or promo timing': ['holiday', 'launch', 'promo', 'campaign'],
}
def _normalize_input(user_input: ReturnInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['product_context', 'review_mode', 'signals', 'reasons', 'notes', 'constraints']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _match_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
for label, keywords in rules.items():
if any(keyword in text for keyword in keywords):
return label
return default
def _match_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
found = [label for label, keywords in rules.items() if any(keyword in text for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class ReturnReasonMiner:
def __init__(self, user_input: ReturnInput):
self.raw = user_input
self.text = _normalize_input(user_input)
self.lower = self.text.lower()
self.review_mode = _match_one(self.lower, REVIEW_MODE_RULES, 'Weekly Return Review')
self.product_context = _match_one(self.lower, PRODUCT_CONTEXT_RULES, 'Apparel')
self.reason_clusters = _match_many(self.lower, REASON_RULES, ['Expectation mismatch', 'Quality defect'], limit=4)
self.signals = _match_many(self.lower, SIGNAL_RULES, ['Return notes and free text', 'SKU or variant clues'], limit=4)
def _taxonomy_rows(self) -> List[List[str]]:
notes = {
'Quality defect': ['Quality defect', 'Product or material issue', 'Review production quality, QC escapes, and vendor consistency.'],
'Size or fit mismatch': ['Size or fit mismatch', 'Merchandising, spec, or expectation issue', 'Inspect size chart clarity, fit guidance, and variant labeling.'],
'Expectation mismatch': ['Expectation mismatch', 'Listing, imagery, or messaging issue', 'Check whether claims, imagery, or use-case framing are overselling the product.'],
'Shipping damage': ['Shipping damage', 'Packaging or carrier issue', 'Inspect packaging protection, handoff points, and damage concentration by route.'],
'Wrong item or fulfillment error': ['Wrong item or fulfillment error', 'Warehouse or pick-pack issue', 'Audit SKU mapping, labeling, and process controls.'],
'Late delivery or timing miss': ['Late delivery or timing miss', 'Carrier, planning, or promise-setting issue', 'Compare promised delivery windows with real operational capability.'],
}
return [notes[item] for item in self.reason_clusters if item in notes]
def render(self) -> str:
lines: List[str] = []
lines.append('# Return Reason Mining Brief')
lines.append('')
lines.append(f'**Review mode:** {self.review_mode}')
lines.append(f'**Product context:** {self.product_context}')
lines.append(f'**Priority reason clusters:** {_join(self.reason_clusters)}')
lines.append(f'**Signals referenced:** {_join(self.signals)}')
lines.append('**Method note:** This is a heuristic return-analysis brief. No live OMS, WMS, ERP, or return portal was accessed.')
lines.append('')
lines.append('## Return Pattern Summary')
lines.append('- Separate product truth problems from fulfillment and expectation-management problems before choosing a fix.')
lines.append('- Recurring returns matter more than memorable one-off anecdotes.')
lines.append(f'- Because the main review mode is **{self.review_mode.lower()}**, the team should prioritize fixes that reduce preventable repeat returns.')
lines.append('')
lines.append('## Reason Taxonomy')
lines.append('| Reason cluster | Likely owner lane | First inspection path |')
lines.append('|---|---|---|')
for row in self._taxonomy_rows():
lines.append(f'| {row[0]} | {row[1]} | {row[2]} |')
lines.append('')
lines.append('## Root Cause Hypotheses')
lines.append('- If quality defect signals are rising, inspect whether the issue is concentrated by batch, vendor, or recent process change.')
lines.append('- If expectation mismatch dominates, inspect imagery, claims, and what the customer thought they were buying.')
lines.append('- If fulfillment or delivery reasons are rising, isolate whether the issue began with warehouse accuracy, packaging, or promise-setting.')
lines.append('')
lines.append('## Fix Priorities')
lines.append('1. Fix the preventable causes with the clearest recurrence and the simplest controllable owner.')
lines.append('2. Improve listing clarity before redesigning the product when the problem is mostly expectation mismatch.')
lines.append('3. Escalate vendor or packaging issues only after confirming the return notes point to the same pattern repeatedly.')
lines.append('4. Track whether the same reason cluster returns after the first remediation cycle.')
lines.append('')
lines.append('## Cross-Functional Action Plan')
lines.append('- **Product:** inspect spec, quality, and packaging assumptions.')
lines.append('- **Merchandising / content:** tighten imagery, size guidance, claims, and comparison framing.')
lines.append('- **Operations:** review pick-pack accuracy, shipping protection, and delivery promise discipline.')
lines.append('- **CX:** standardize how return reasons are captured so future analysis is cleaner.')
lines.append('')
lines.append('## Assumptions and Limits')
lines.append('- This brief is heuristic and depends on the return notes and context supplied by the user.')
lines.append('- Root causes are hypotheses until validated by repeated evidence or operational review.')
lines.append('- Final product, vendor, inventory, refund, and policy decisions remain human-approved.')
return "\n".join(lines)
def handle(user_input: ReturnInput) -> str:
return ReturnReasonMiner(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import ReturnReasonMiner, handle
def test_review_mode_detection_launch():
miner = ReturnReasonMiner('We need launch issue triage for our new skincare product because returns are rising.')
assert miner.review_mode == 'Launch Issue Triage'
def test_product_context_and_reason_detection_apparel_fit():
miner = ReturnReasonMiner('Apparel returns mention size too small and fit mismatch across several variants.')
assert miner.product_context == 'Apparel'
assert 'Size or fit mismatch' in miner.reason_clusters
def test_reason_detection_damage_and_wrong_item():
miner = ReturnReasonMiner('Customers report damaged packages and wrong items from the warehouse.')
assert 'Shipping damage' in miner.reason_clusters
assert 'Wrong item or fulfillment error' in miner.reason_clusters
def test_render_contains_sections():
output = handle('Create a weekly return review for electronics returns and refund notes.')
assert output.startswith('# Return Reason Mining Brief')
assert '## Reason Taxonomy' in output
assert '## Fix Priorities' in output
assert '## Cross-Functional Action Plan' in output
def test_dict_input_supported():
output = handle({
'product_context': 'beauty',
'review_mode': 'weekly return review',
'reasons': ['damaged', 'not as described'],
'signals': ['review', 'packaging'],
})
assert '# Return Reason Mining Brief' in output
assert 'Beauty / Personal Care' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Review support quality for chat, email, phone, social DM, or marketplace service teams. Use when a team needs a QA scorecard, failure-mode analysis, coaching...
---
name: service-qa-coach
description: Review support quality for chat, email, phone, social DM, or marketplace service teams. Use when a team needs a QA scorecard, failure-mode analysis, coaching plan, calibration checklist, or service-improvement brief without live ticketing, CRM, or contact-center system access.
---
# Service QA Coach
## Overview
Use this skill to turn support quality concerns into a structured QA and coaching brief. It helps define the scorecard, isolate likely failure modes, recommend coaching actions, and organize a sampling and calibration rhythm.
This MVP is heuristic. It does **not** connect to live ticketing systems, call recordings, chat exports, CRMs, or QA platforms. It relies on the user's provided support notes, issue patterns, and service expectations.
## Trigger
Use this skill when the user wants to:
- audit customer service quality across chat, email, phone, social DM, or marketplace tickets
- build or refine a QA rubric for support teams
- diagnose recurring service failures such as poor empathy, policy errors, or weak resolution quality
- create a coaching plan for agents, team leads, or new hires
- run a calibration sprint before changing SOPs or scorecards
### Example prompts
- "Help me build a QA scorecard for our live chat team"
- "Why are refund escalations rising in customer service?"
- "Create a coaching plan for agents with low empathy and policy accuracy"
- "Turn these support issues into a QA and calibration brief"
## Workflow
1. Capture the support channel, review mode, and quality concern.
2. Choose the focus areas such as compliance, empathy, resolution, speed, or documentation.
3. Identify the likely failure modes and coaching priorities.
4. Define sampling, calibration, and follow-up rhythm.
5. Return a markdown QA coaching brief.
## Inputs
The user can provide any mix of:
- support channel such as chat, email, phone, social DM, or marketplace tickets
- quality goals such as CSAT recovery, policy compliance, or escalation reduction
- issue notes such as slow responses, low empathy, incorrect policy answers, or poor note-taking
- team context such as new hires, macro usage, QA backlog, or uneven supervisor coaching
- sample size, review cadence, and escalation process assumptions
## Outputs
Return a markdown QA coaching brief with:
- QA summary
- scorecard rubric
- failure mode review
- coaching plan
- calibration and sampling guidance
- assumptions and limits
## Safety
- Do not claim access to live tickets, recordings, or customer records.
- Do not invent agent-level performance facts that were not supplied.
- Privacy, compliance, refund, and disciplinary decisions remain human-approved.
- Reduce certainty when the sample is tiny or anecdotal.
## Best-fit Scenarios
- service teams that need a lighter QA framework before investing in tooling
- ecommerce operators managing outsourced or mixed support teams
- managers trying to reduce escalations, refunds, or poor customer sentiment
## Not Ideal For
- live contact-center monitoring or real-time QA enforcement
- regulated complaint handling that requires legal or compliance review
- detailed workforce management or call-center forecasting
## Acceptance Criteria
- Return markdown text.
- Include rubric, coaching, calibration, and limits sections.
- Keep the no-live-data framing explicit.
- Make the output practical for team leads and support managers.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Sequence, Union
QaInput = Union[str, Dict[str, Any]]
REVIEW_MODE_RULES = {
'Escalation Reduction': ['escalation', 'refund', 'angry customer', 'complaint'],
'CSAT Recovery': ['csat', 'satisfaction', 'low score', 'bad feedback'],
'Marketplace Support Audit': ['marketplace', 'amazon buyer message', 'shopify inbox'],
'New Hire Ramp': ['new hire', 'onboarding', 'ramp'],
'Calibration Sprint': ['calibration', 'scorecard', 'rubric'],
}
CHANNEL_RULES = {
'Live chat': ['chat', 'live chat'],
'Email support': ['email'],
'Phone support': ['phone', 'call'],
'Social DM': ['dm', 'social', 'instagram', 'facebook', 'wechat'],
'Marketplace tickets': ['marketplace', 'amazon', 'ticket'],
}
FOCUS_RULES = {
'Policy compliance': ['policy', 'compliance', 'refund policy'],
'Empathy and tone': ['empathy', 'tone', 'rude', 'robotic'],
'Resolution quality': ['resolve', 'resolution', 'solved', 'fix'],
'Response speed': ['slow', 'sla', 'response time', 'delay'],
'Documentation accuracy': ['notes', 'documentation', 'case note', 'logging'],
}
ISSUE_RULES = {
'Slow first response': ['slow', 'delay', 'waiting'],
'Low empathy / robotic tone': ['robotic', 'cold', 'rude', 'empathy'],
'Policy misquote': ['wrong policy', 'policy', 'incorrect refund'],
'Incomplete troubleshooting': ['did not solve', 'not solved', 'troubleshoot', 'incomplete'],
'Excessive escalation or transfer': ['escalation', 'transfer', 'handoff'],
'Weak case notes': ['notes', 'logging', 'documentation'],
}
def _normalize_input(user_input: QaInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['channel', 'goal', 'issues', 'team_context', 'sample', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _match_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
for label, keywords in rules.items():
if any(keyword in text for keyword in keywords):
return label
return default
def _match_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
found = [label for label, keywords in rules.items() if any(keyword in text for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class ServiceQaCoach:
def __init__(self, user_input: QaInput):
self.raw = user_input
self.text = _normalize_input(user_input)
self.lower = self.text.lower()
self.review_mode = _match_one(self.lower, REVIEW_MODE_RULES, 'Calibration Sprint')
self.channels = _match_many(self.lower, CHANNEL_RULES, ['Live chat', 'Email support'], limit=4)
self.focus_areas = _match_many(self.lower, FOCUS_RULES, ['Empathy and tone', 'Resolution quality'], limit=4)
self.issue_flags = _match_many(self.lower, ISSUE_RULES, ['Slow first response', 'Low empathy / robotic tone'], limit=4)
def _rubric_rows(self) -> List[List[str]]:
rows = {
'Policy compliance': ['Policy compliance', 'Was the answer accurate, safe, and aligned to policy?', 'Incorrect policy guidance creates risk fast.'],
'Empathy and tone': ['Empathy and tone', 'Did the customer feel understood without losing clarity?', 'Cold or robotic tone can escalate frustration even when the answer is technically correct.'],
'Resolution quality': ['Resolution quality', 'Did the interaction move the case materially closer to a solution?', 'Fast replies that do not solve the problem still create repeat contact.'],
'Response speed': ['Response speed', 'Was the timing appropriate for the channel and case severity?', 'Speed matters, but not more than accuracy and clarity.'],
'Documentation accuracy': ['Documentation accuracy', 'Could another agent pick up the case cleanly from the notes?', 'Weak notes drive repeat work and messy escalations.'],
}
return [rows[item] for item in self.focus_areas if item in rows]
def render(self) -> str:
lines: List[str] = []
lines.append('# Service QA Coaching Brief')
lines.append('')
lines.append(f'**Review mode:** {self.review_mode}')
lines.append(f'**Support channels:** {_join(self.channels)}')
lines.append(f'**Focus areas:** {_join(self.focus_areas)}')
lines.append(f'**Issue flags:** {_join(self.issue_flags)}')
lines.append('**Method note:** This is a heuristic QA brief. No live ticketing, CRM, call, or chat system was accessed.')
lines.append('')
lines.append('## QA Summary')
lines.append('- Separate whether the team has a knowledge problem, a behavior problem, or an operating-rhythm problem.')
lines.append('- Score quality on what helps the customer move forward, not on script compliance alone.')
lines.append(f'- Because the main review mode is **{self.review_mode.lower()}**, the plan should focus on repeatable coaching rather than one-off criticism.')
lines.append('')
lines.append('## Scorecard Rubric')
lines.append('| Dimension | Core question | Coaching angle |')
lines.append('|---|---|---|')
for row in self._rubric_rows():
lines.append(f'| {row[0]} | {row[1]} | {row[2]} |')
lines.append('')
lines.append('## Failure Mode Review')
for item in self.issue_flags:
lines.append(f'- **{item}:** isolate where the workflow, training, or judgment broke instead of blaming the agent generically.')
lines.append('')
lines.append('## Coaching Plan')
lines.append('1. Review a small but representative sample and tag every miss by failure mode, not by gut feel alone.')
lines.append('2. Turn the top two recurring misses into short coaching moments with examples of what good looks like.')
lines.append('3. Recheck the same agent or queue quickly after coaching so improvement is measured, not assumed.')
lines.append('4. Escalate process or policy gaps separately from individual coaching needs.')
lines.append('')
lines.append('## Calibration and Sampling Plan')
lines.append('- Use the same sample definition across reviewers so scores are comparable.')
lines.append('- Calibrate on edge cases first, especially refunds, angry customers, or unclear policy situations.')
lines.append('- Keep one running list of rubric disagreements and resolve them before changing score weights.')
lines.append('')
lines.append('## Assumptions and Limits')
lines.append('- This brief is heuristic and depends on the notes, excerpts, and issue framing supplied by the user.')
lines.append('- Privacy, compliance, HR, refund, and disciplinary decisions remain human-approved.')
lines.append('- Tiny or anecdotal samples should be treated as signals to inspect further, not as proof of team-wide performance.')
return "\n".join(lines)
def handle(user_input: QaInput) -> str:
return ServiceQaCoach(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import ServiceQaCoach, handle
def test_review_mode_detection_escalation():
coach = ServiceQaCoach('Refund escalations are rising and we need an escalation reduction plan.')
assert coach.review_mode == 'Escalation Reduction'
def test_channel_detection_chat_email():
coach = ServiceQaCoach('Audit our live chat and email service quality this month.')
assert 'Live chat' in coach.channels
assert 'Email support' in coach.channels
def test_focus_and_issue_detection():
coach = ServiceQaCoach('Agents sound robotic, responses are slow, and notes are incomplete.')
assert 'Empathy and tone' in coach.focus_areas
assert 'Response speed' in coach.focus_areas
assert 'Weak case notes' in coach.issue_flags
def test_render_contains_sections():
output = handle('Build a QA scorecard and coaching plan for our support team.')
assert output.startswith('# Service QA Coaching Brief')
assert '## Scorecard Rubric' in output
assert '## Coaching Plan' in output
assert '## Calibration and Sampling Plan' in output
def test_dict_input_supported():
output = handle({
'channel': ['chat', 'marketplace'],
'goal': 'csat recovery',
'issues': ['slow response', 'wrong policy'],
'team_context': 'new hire queue',
})
assert '# Service QA Coaching Brief' in output
assert 'CSAT Recovery' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Build a lightweight competitor monitoring brief across DTC sites, marketplaces, social channels, and promo surfaces. Use when a team needs pricing or promo s...
---
name: competitor-watchtower
description: Build a lightweight competitor monitoring brief across DTC sites, marketplaces, social channels, and promo surfaces. Use when a team needs pricing or promo scans, assortment watchlists, messaging comparison, threat assessment, or response options without live scraping, BI, or market-intelligence platform access.
---
# Competitor Watchtower
## Overview
Use this skill to turn scattered competitor observations into a structured watch brief. It helps operators focus on the signals that matter most, frame the likely threat or opportunity, and decide what to monitor, ignore, or respond to next.
This MVP is heuristic. It does **not** connect to live price trackers, scraping tools, ad libraries, marketplace APIs, or analytics systems. It relies on the user's supplied competitor notes, channel context, and business priorities.
## Trigger
Use this skill when the user wants to:
- monitor competitor pricing, promo pressure, launches, or assortment moves
- compare competitor messaging, proof strategy, or service promises
- build a weekly watchlist for ecommerce or marketplace operators
- convert rough screenshots or notes into a structured competitor brief
- decide whether a competitor move needs a response now, later, or not at all
### Example prompts
- "Help me build a weekly competitor watchlist for Amazon and TikTok Shop"
- "A rival brand is discounting hard. How should we frame the threat?"
- "Turn these pricing and launch notes into a competitor brief"
- "What should we monitor when a premium competitor launches a new bundle?"
## Workflow
1. Capture the watch purpose, market surface, and named competitors if available.
2. Choose the likely watch mode, such as pricing, launch, creative, or service scan.
3. Organize the highest-signal moves into a threat and opportunity view.
4. Suggest response paths, watch cadence, and owner roles.
5. Return a markdown competitor brief with assumptions and limits.
## Inputs
The user can provide any mix of:
- channels or surfaces such as DTC sites, Amazon, Tmall, TikTok Shop, Xiaohongshu, or retail
- competitor notes about price, discounts, bundles, reviews, assortment, shipping, or messaging
- strategic context such as launch period, promo window, margin pressure, or brand positioning
- internal priorities such as price integrity, share defense, new customer acquisition, or category expansion
- evidence quality notes such as screenshots, anecdotal field notes, or incomplete observations
## Outputs
Return a markdown competitor brief with:
- competitive situation summary
- signal grid by watch category
- threat and opportunity notes
- response options and watch cadence
- owner hints and assumption notes
## Safety
- Do not claim access to live scraped data or market-intelligence platforms.
- Treat competitor interpretation as directional unless the evidence is strong and recent.
- Do not recommend deceptive, anti-competitive, or policy-violating actions.
- Final pricing, assortment, and legal decisions remain human-approved.
## Best-fit Scenarios
- ecommerce teams that need a practical weekly or campaign-based competitor scan
- marketplace sellers facing visible price or promo pressure
- founders or operators who need structure before reacting to a rival move
## Not Ideal For
- real-time scraping or automated alerting infrastructure
- legal claims about competitor misconduct without verified evidence
- formal market sizing or investor-grade intelligence work
## Acceptance Criteria
- Return markdown text.
- Include signal grid, response options, and limits sections.
- Keep the advisory and no-live-data framing explicit.
- Make the brief useful for commercial operators, not just analysts.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Sequence, Union
WatchInput = Union[str, Dict[str, Any]]
WATCH_MODE_RULES = {
'Pricing and Promo Scan': ['price', 'pricing', 'discount', 'promo', 'promotion', 'coupon', 'markdown'],
'Assortment and Launch Scan': ['launch', 'new sku', 'new product', 'assortment', 'variant', 'bundle'],
'Creative and Messaging Scan': ['creative', 'message', 'copy', 'ugc', 'ad', 'content'],
'Service and CX Scan': ['service', 'support', 'shipping', 'delivery', 'refund'],
'Weekly General Watch': ['weekly', 'monitor', 'watchlist', 'watch'],
}
SURFACE_RULES = {
'DTC site': ['dtc', 'site', 'website', 'shopify'],
'Amazon': ['amazon'],
'Tmall / Taobao': ['tmall', 'taobao', '淘宝'],
'TikTok Shop / Douyin': ['tiktok', 'douyin'],
'Xiaohongshu': ['xiaohongshu', 'xhs', 'rednote'],
'Retail / offline': ['retail', 'store shelf', 'offline'],
}
SIGNAL_RULES = {
'Price and discounting': ['price', 'discount', 'coupon', 'promo', 'markdown'],
'Assortment and launch': ['launch', 'sku', 'assortment', 'variant', 'bundle'],
'Reviews and proof': ['review', 'rating', 'testimonial', 'proof'],
'Creative and messaging': ['creative', 'message', 'copy', 'ugc', 'ad'],
'Service and fulfillment': ['shipping', 'delivery', 'refund', 'service', 'support'],
}
PRESSURE_RULES = {
'Aggressive price pressure': ['heavy discount', 'aggressive', 'deep discount', 'price war'],
'Launch visibility risk': ['launch', 'new product', 'debut'],
'Trust or proof drift': ['review', 'rating', 'testimonial', 'ugc'],
'Service perception risk': ['late delivery', 'refund issues', 'support complaints'],
}
def _normalize_input(user_input: WatchInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['surfaces', 'competitors', 'watch_mode', 'signals', 'notes', 'priority']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _match_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
for label, keywords in rules.items():
if any(keyword in text for keyword in keywords):
return label
return default
def _match_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
found = [label for label, keywords in rules.items() if any(keyword in text for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class CompetitorWatchtower:
def __init__(self, user_input: WatchInput):
self.raw = user_input
self.text = _normalize_input(user_input)
self.lower = self.text.lower()
self.watch_mode = _match_one(self.lower, WATCH_MODE_RULES, 'Weekly General Watch')
self.surfaces = _match_many(self.lower, SURFACE_RULES, ['DTC site', 'Amazon'], limit=4)
self.signals = _match_many(self.lower, SIGNAL_RULES, ['Price and discounting', 'Assortment and launch'], limit=4)
self.pressure_flags = _match_many(self.lower, PRESSURE_RULES, ['Aggressive price pressure'], limit=3)
def _signal_rows(self) -> List[List[str]]:
notes = {
'Price and discounting': ['Price and discounting', 'Track list price, bundle logic, discount depth, and whether urgency is sustainable.'],
'Assortment and launch': ['Assortment and launch', 'Watch for hero SKU changes, new bundles, premiumization, or defensive line extensions.'],
'Reviews and proof': ['Reviews and proof', 'Check whether the competitor is strengthening trust with ratings, UGC, or comparison claims.'],
'Creative and messaging': ['Creative and messaging', 'Look for sharper hooks, clearer proof, or positioning shifts that could change buyer perception.'],
'Service and fulfillment': ['Service and fulfillment', 'Monitor delivery promises, refund friction, and support positioning that can alter conversion quality.'],
}
return [notes[item] for item in self.signals if item in notes]
def render(self) -> str:
lines: List[str] = []
lines.append('# Competitor Watchtower Brief')
lines.append('')
lines.append(f'**Watch mode:** {self.watch_mode}')
lines.append(f'**Surfaces in scope:** {_join(self.surfaces)}')
lines.append(f'**Priority signals:** {_join(self.signals)}')
lines.append(f'**Pressure flags:** {_join(self.pressure_flags)}')
lines.append('**Method note:** This is a heuristic competitor brief. No live scraping, ad library, marketplace API, or BI platform was accessed.')
lines.append('')
lines.append('## Competitive Situation Summary')
lines.append('- Treat competitor movement as a signal to interpret, not a command to react instantly.')
lines.append('- Ask whether the rival move threatens your margin, conversion path, trust position, or category narrative.')
lines.append(f'- Because the main watch mode is **{self.watch_mode.lower()}**, the operator should separate real commercial risk from performative noise.')
lines.append('')
lines.append('## Signal Grid')
lines.append('| Signal | What to inspect first |')
lines.append('|---|---|')
for row in self._signal_rows():
lines.append(f'| {row[0]} | {row[1]} |')
lines.append('')
lines.append('## Threats and Opportunity Windows')
lines.append('- If the competitor is discounting hard, decide whether the threat is real share loss or just visible noise with weak staying power.')
lines.append('- If a rival launched something new, check whether it fills a real whitespace or simply reframes an existing offer.')
lines.append('- Look for opportunities where the competitor is teaching the market but leaving trust, service, or value gaps open.')
lines.append('')
lines.append('## Response Options')
lines.append('1. Hold position when the rival move does not materially change buyer choice or category economics.')
lines.append('2. Strengthen proof, offer clarity, or merchandising before matching a competitor on price.')
lines.append('3. Escalate only the moves that affect your hero SKUs, margin guardrails, or launch window.')
lines.append('4. Create a tighter follow-up watchlist for any unresolved threat instead of reacting with broad panic.')
lines.append('')
lines.append('## Watch Cadence and Owners')
lines.append('- **Weekly operator scan:** price, promo, and hero-SKU changes.')
lines.append('- **Campaign window scan:** creative, launch, and landing-page changes during high-risk periods.')
lines.append('- **Owner suggestion:** one commercial owner, one marketplace owner, and one brand or content owner for interpretation.')
lines.append('')
lines.append('## Assumptions and Limits')
lines.append('- This brief depends on the evidence quality supplied by the user and may overstate or understate competitor intent when notes are incomplete.')
lines.append('- Legal, pricing, assortment, and public-response decisions remain human-approved.')
lines.append('- Avoid overfitting to one screenshot, one campaign, or one temporary discount burst.')
return "\n".join(lines)
def handle(user_input: WatchInput) -> str:
return CompetitorWatchtower(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import CompetitorWatchtower, handle
def test_watch_mode_detection_pricing():
tower = CompetitorWatchtower('Run a pricing scan because the competitor launched heavy discount coupons.')
assert tower.watch_mode == 'Pricing and Promo Scan'
def test_surface_detection_amazon_tiktok():
tower = CompetitorWatchtower('Monitor Amazon and TikTok Shop competitors this week.')
assert 'Amazon' in tower.surfaces
assert 'TikTok Shop / Douyin' in tower.surfaces
def test_signal_detection_reviews_and_service():
tower = CompetitorWatchtower('Watch reviews, ratings, refunds, and shipping complaints from the rival brand.')
assert 'Reviews and proof' in tower.signals
assert 'Service and fulfillment' in tower.signals
def test_render_contains_sections():
output = handle('Build a weekly competitor watchlist for Amazon and our DTC site.')
assert output.startswith('# Competitor Watchtower Brief')
assert '## Signal Grid' in output
assert '## Response Options' in output
assert '## Assumptions and Limits' in output
def test_dict_input_supported():
output = handle({
'surfaces': ['Amazon', 'DTC site'],
'watch_mode': 'pricing and promo scan',
'signals': ['price', 'reviews'],
'notes': 'competitor has aggressive discounting',
})
assert '# Competitor Watchtower Brief' in output
assert 'Pricing and Promo Scan' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Plan creator partnerships across TikTok, Douyin, Xiaohongshu, Instagram, YouTube, or similar channels. Use when a team needs creator mix recommendations, out...
---
name: creator-collaboration-planner
description: Plan creator partnerships across TikTok, Douyin, Xiaohongshu, Instagram, YouTube, or similar channels. Use when a team needs creator mix recommendations, outreach logic, brief structure, compensation options, collaboration timeline, or risk controls without live creator marketplace or attribution platform access.
---
# Creator Collaboration Planner
## Overview
Use this skill to turn a vague creator marketing goal into a collaboration plan that the operator can actually run. It helps define creator mix, selection criteria, outreach sequence, compensation structure, brief essentials, and measurement logic.
This MVP is heuristic. It does **not** connect to live creator databases, affiliate platforms, social analytics tools, or contract systems. It relies on the user's provided campaign context, creator assumptions, and operating constraints.
## Trigger
Use this skill when the user wants to:
- build a creator collaboration plan for a launch, seasonal push, or always-on program
- choose between seeding, paid, affiliate, ambassador, or live selling partnerships
- brief creators with the right message, proof points, and deliverable expectations
- structure creator tiers, compensation, negotiation guardrails, and review cadence
- reduce creator program risk before outreach starts
### Example prompts
- "Help me build a creator plan for our new product launch on Xiaohongshu and TikTok"
- "What kind of creators should we recruit for seeding versus affiliate sales?"
- "Create a creator outreach and brief plan for a beauty campaign"
- "How should we structure fees, gifting, and tracking for micro creators?"
## Workflow
1. Capture the commercial goal, target channels, and collaboration format.
2. Choose the likely program type, creator mix, and compensation logic.
3. Define the brief, outreach sequence, and approval workflow.
4. Map risks such as misalignment, disclosure issues, and slow approvals.
5. Return a markdown collaboration plan with timeline and debrief guidance.
## Inputs
The user can provide any mix of:
- target channels such as TikTok, Douyin, Xiaohongshu, Instagram, YouTube, or Bilibili
- campaign goal such as awareness, conversion, review seeding, or launch momentum
- creator preferences such as micro creators, UGC makers, experts, affiliates, or live hosts
- deliverables, fee assumptions, gifting plan, commission logic, and approval constraints
- timeline, product samples, usage rules, and disclosure requirements
- known risks such as vague brief quality, tracking gaps, or slow legal approval
## Outputs
Return a markdown collaboration plan with:
- collaboration strategy summary
- creator mix and selection rubric
- offer structure and brief essentials
- outreach and negotiation plan
- timeline and operating rhythm
- measurement and debrief guidance
- risk controls and limitations
## Safety
- Do not claim access to live creator rosters, audience analytics, or attribution tools.
- Do not promise reach, sales, or ROAS outcomes from a creator plan.
- Compensation, legal review, and disclosure decisions remain human-approved.
- Reduce certainty when audience fit or product differentiation is weak.
## Best-fit Scenarios
- brands building creator programs without an in-house influencer lead
- operators who need a practical seeding or paid-collaboration framework
- teams that want clearer creator selection and brief quality before spending
## Not Ideal For
- automated creator recruitment or in-app campaign execution
- final legal contract drafting for regulated or highly sensitive categories
- attribution modeling that depends on live analytics or clean tracking infrastructure
## Acceptance Criteria
- Return markdown text.
- Include creator mix, brief, outreach, timeline, and risk sections.
- Keep the no-live-data limitation explicit.
- Make the output practical for a brand or ecommerce operator.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Sequence, Union
PlanInput = Union[str, Dict[str, Any]]
PROGRAM_RULES = {
'Product Seeding Program': ['seed', 'seeding', 'sample', 'gift'],
'Paid Campaign Brief': ['paid', 'fee', 'sponsored', 'deliverable', 'whitelist'],
'Affiliate / Commission Program': ['affiliate', 'commission', 'cps', 'rev share'],
'Launch Squad / Ambassador Program': ['ambassador', 'always-on', 'community', 'long-term'],
'Live Selling Collaboration': ['live', 'livestream', 'stream', 'co-host'],
}
CHANNEL_RULES = {
'TikTok / Douyin': ['tiktok', 'douyin'],
'Xiaohongshu': ['xiaohongshu', 'xhs', 'rednote'],
'Instagram': ['instagram'],
'YouTube': ['youtube'],
'Bilibili': ['bilibili'],
'WeChat': ['wechat'],
}
OBJECTIVE_RULES = {
'Awareness': ['awareness', 'reach', 'visibility'],
'Conversion': ['conversion', 'sales', 'purchase', 'affiliate'],
'Review Seeding': ['review', 'seed', 'ugc', 'testimonial'],
'Launch Momentum': ['launch', 'drop', 'debut'],
'Content Library Build': ['content library', 'ugc', 'asset', 'creative'],
}
CREATOR_TYPE_RULES = {
'UGC creators': ['ugc', 'content creator'],
'Micro creators': ['micro', 'nano', 'small creator'],
'Expert creators': ['expert', 'doctor', 'trainer', 'specialist', 'authority'],
'Affiliate creators': ['affiliate', 'commission', 'deal'],
'Live hosts': ['live', 'host', 'stream'],
}
RISK_RULES = {
'Audience mismatch': ['wrong audience', 'misaligned', 'fit'],
'Brief ambiguity': ['brief', 'unclear', 'vague', 'confusing'],
'Tracking gap': ['tracking', 'attribution', 'link', 'coupon code'],
'Disclosure / compliance risk': ['disclosure', 'compliance', 'ad label', 'policy'],
'Slow approval cycle': ['approval', 'legal', 'review delay'],
'Overpaying for weak proof': ['high fee', 'expensive', 'costly', 'overpay'],
}
def _normalize_input(user_input: PlanInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['channels', 'objective', 'creator_types', 'program', 'deliverables', 'risks', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _match_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
for label, keywords in rules.items():
if any(keyword in text for keyword in keywords):
return label
return default
def _match_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
found = [label for label, keywords in rules.items() if any(keyword in text for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class CreatorCollaborationPlanner:
def __init__(self, user_input: PlanInput):
self.raw = user_input
self.text = _normalize_input(user_input)
self.lower = self.text.lower()
self.program_type = _match_one(self.lower, PROGRAM_RULES, 'Product Seeding Program')
self.channels = _match_many(self.lower, CHANNEL_RULES, ['TikTok / Douyin', 'Xiaohongshu'], limit=4)
self.objective = _match_one(self.lower, OBJECTIVE_RULES, 'Awareness')
self.creator_mix = _match_many(self.lower, CREATOR_TYPE_RULES, ['Micro creators', 'UGC creators'], limit=4)
self.risks = _match_many(self.lower, RISK_RULES, ['Brief ambiguity', 'Tracking gap'], limit=4)
def _rubric_rows(self) -> List[List[str]]:
base = {
'UGC creators': ['UGC creators', 'Fast asset generation and social proof', 'Weak differentiation if the brief is generic'],
'Micro creators': ['Micro creators', 'Affordable reach with stronger community trust', 'Results depend on fit and follow-through, not follower count alone'],
'Expert creators': ['Expert creators', 'Credibility for high-consideration or trust-sensitive products', 'Require tighter fact control and usually slower approval'],
'Affiliate creators': ['Affiliate creators', 'Performance-minded distribution with sales accountability', 'Needs cleaner tracking, incentive logic, and margin discipline'],
'Live hosts': ['Live hosts', 'High-intent commerce moments and interactive selling', 'Operational load is higher and the brief must be much tighter'],
}
return [base[item] for item in self.creator_mix if item in base]
def render(self) -> str:
lines: List[str] = []
lines.append('# Creator Collaboration Plan')
lines.append('')
lines.append(f'**Program type:** {self.program_type}')
lines.append(f'**Channels in scope:** {_join(self.channels)}')
lines.append(f'**Primary objective:** {self.objective}')
lines.append(f'**Creator mix:** {_join(self.creator_mix)}')
lines.append(f'**Risk flags:** {_join(self.risks)}')
lines.append('**Method note:** This is a heuristic collaboration plan. No live creator database, audience analytics, or attribution tool was accessed.')
lines.append('')
lines.append('## Collaboration Strategy Summary')
lines.append('- Choose the collaboration model based on what the business actually needs: trust, content volume, reach, or conversion proof.')
lines.append(f'- Because the main objective is **{self.objective.lower()}**, the creator plan should optimize for fit and message clarity before scale.')
lines.append('- Use one operator-owned source of truth for creator status, samples, approvals, and content rights.')
lines.append('')
lines.append('## Creator Mix and Selection Rubric')
lines.append('| Creator type | Best use | Main watch-out |')
lines.append('|---|---|---|')
for row in self._rubric_rows():
lines.append(f'| {row[0]} | {row[1]} | {row[2]} |')
lines.append('')
lines.append('## Offer Structure and Brief Essentials')
lines.append('- Define what the creator must communicate, what proof is mandatory, and what they must avoid improvising.')
lines.append('- Keep the compensation logic simple enough that the creator can understand how value is created on both sides.')
lines.append('- Clarify deliverables, approval rounds, usage rights, posting window, and disclosure expectations before sending samples or contracts.')
lines.append('')
lines.append('## Outreach and Negotiation Plan')
lines.append('1. Build a short target list by creator fit, not by vanity reach alone.')
lines.append('2. Send outreach with one clear hook, one product reason to care, and one simple ask.')
lines.append('3. Qualify creators on audience fit, execution quality, speed, and willingness to follow the brief.')
lines.append('4. Negotiate deliverables, usage rights, timing, and payment logic before the product ships or the brief goes live.')
lines.append('')
lines.append('## Timeline and Operating Rhythm')
lines.append('- **Week 1:** finalize goal, shortlist creators, and lock brief essentials.')
lines.append('- **Week 2:** outreach, negotiation, sample logistics, and disclosure prep.')
lines.append('- **Week 3:** creator content production, review, and revision management.')
lines.append('- **Week 4:** publishing, tracking, debrief, and next-wave decisions.')
lines.append('')
lines.append('## Measurement and Debrief')
lines.append('- Track fit-to-output first: response rate, acceptance rate, content quality, and revision burden.')
lines.append('- Then track campaign effect using the best available evidence such as code usage, click quality, saves, comments, or assisted conversions.')
lines.append('- Capture which creator archetypes generated the strongest trust, content reusability, and commercial signal.')
lines.append('')
lines.append('## Risk Controls and Limits')
lines.append('- This plan is heuristic and depends on the user-supplied audience, product, and channel context.')
lines.append('- Legal, brand safety, disclosure, payment, and rights decisions remain human-approved.')
lines.append('- Do not confuse creator popularity with creator fit or sales quality.')
return "\n".join(lines)
def handle(user_input: PlanInput) -> str:
return CreatorCollaborationPlanner(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import CreatorCollaborationPlanner, handle
def test_channel_detection_tiktok_xiaohongshu():
planner = CreatorCollaborationPlanner('Build a creator plan for TikTok and Xiaohongshu.')
assert 'TikTok / Douyin' in planner.channels
assert 'Xiaohongshu' in planner.channels
def test_program_type_detection_affiliate():
planner = CreatorCollaborationPlanner('We want an affiliate program with commission payouts for creators.')
assert planner.program_type == 'Affiliate / Commission Program'
def test_creator_mix_detection_expert_and_ugc():
planner = CreatorCollaborationPlanner('Need expert creators plus UGC content creators for a trust-building beauty launch.')
assert 'Expert creators' in planner.creator_mix
assert 'UGC creators' in planner.creator_mix
def test_render_contains_sections():
output = handle('Create a creator outreach and brief plan for a product launch on Xiaohongshu.')
assert output.startswith('# Creator Collaboration Plan')
assert '## Creator Mix and Selection Rubric' in output
assert '## Outreach and Negotiation Plan' in output
assert '## Measurement and Debrief' in output
def test_dict_input_supported():
output = handle({
'channels': ['TikTok', 'Instagram'],
'objective': 'awareness',
'creator_types': ['micro creators', 'ugc'],
'program': 'paid sponsored deliverables',
})
assert '# Creator Collaboration Plan' in output
assert 'Paid Campaign Brief' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Plan and run a live commerce session across Douyin, TikTok Shop, Taobao Live, Amazon Live, or similar channels. Use when a team needs a run of show, offer la...
---
name: live-commerce-showrunner
description: Plan and run a live commerce session across Douyin, TikTok Shop, Taobao Live, Amazon Live, or similar channels. Use when a team needs a run of show, offer ladder, host script outline, staffing checklist, moderation plan, risk controls, or post-live debrief structure without live platform or GMV data access.
---
# Live Commerce Showrunner
## Overview
Use this skill to turn a rough live selling idea into an operator-ready show brief. It helps structure the show objective, offer flow, host responsibilities, moderation plays, risk controls, and debrief checklist.
This MVP is heuristic. It does **not** connect to live commerce dashboards, inventory systems, ad platforms, comment feeds, or payment tools. It relies on the user's provided channel context, product priorities, offer assumptions, and team constraints.
## Trigger
Use this skill when the user wants to:
- build a run of show for a live commerce session
- plan a launch, promo, clearance, or education-led livestream
- brief hosts, moderators, operators, and product assistants
- design the offer ladder, urgency moments, and comment-conversion plays
- create a pre-show checklist and post-show debrief template
### Example prompts
- "Help me plan a 45-minute Douyin live for our new skincare launch"
- "Create a run of show for a clearance livestream on TikTok Shop"
- "What should our host, moderator, and ops lead each do during the live?"
- "Build a backup plan for low traffic and inventory risk during a live selling event"
## Workflow
1. Capture the show goal, channel, lead products, and commercial constraints.
2. Choose the likely show type, such as launch, promo, education, or guest session.
3. Build the run of show with opening hook, proof moments, offer stacking, and close.
4. Define host, moderator, ops, and product support responsibilities.
5. Return a markdown show brief with risk controls and debrief guidance.
## Inputs
The user can provide any mix of:
- live channel such as Douyin, TikTok Shop, Taobao Live, Amazon Live, or Instagram Live
- show objective such as launch conversion, stock clearance, education, or audience growth
- product priorities, bundles, pricing, or coupon assumptions
- host profile, guest involvement, moderator support, and crew availability
- planned show length, promo calendar, and traffic expectations
- risk notes such as low stock, compliance pressure, weak script confidence, or technical concerns
## Outputs
Return a markdown show brief with:
- show strategy summary
- run of show by segment
- offer and merch plan
- host and crew checklist
- comment moderation and conversion plays
- backup plan for common failure modes
- debrief checklist with learning questions
## Safety
- Do not claim access to live traffic, GMV, comment, or stock systems.
- Do not promise sales outcomes from the show plan.
- Compliance, pricing, and inventory decisions remain human-approved.
- Downgrade certainty when the user provides weak offer, product, or traffic detail.
## Best-fit Scenarios
- brand or marketplace teams running founder-led or host-led live selling sessions
- operators who need structure before a launch or promo livestream
- smaller teams that do not yet have a dedicated live commerce producer
## Not Ideal For
- live in-session control or real-time comment moderation
- guaranteed forecasting of GMV, traffic, or conversion
- regulated categories that require formal legal review before scripting
## Acceptance Criteria
- Return markdown text.
- Include run of show, team roles, risk controls, and debrief sections.
- Make the no-live-data limitation explicit.
- Keep the brief practical for a host, moderator, and ecommerce operator.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Sequence, Union
ShowInput = Union[str, Dict[str, Any]]
SHOW_TYPE_RULES = {
'Product Launch Live': ['launch', 'new product', 'debut', 'drop'],
'Promo / Clearance Push': ['clearance', 'promo', 'promotion', 'discount', 'markdown', 'flash sale', '618', '11.11'],
'Education / Demo Session': ['demo', 'education', 'tutorial', 'how to use', 'explain'],
'Guest / Co-host Session': ['guest', 'co-host', 'creator', 'expert', 'influencer', 'collab'],
'VIP / Community Session': ['vip', 'member', 'community', 'loyalty', 'private'],
}
CHANNEL_RULES = {
'Douyin Live': ['douyin', '抖音'],
'TikTok Shop Live': ['tiktok', 'tiktok shop'],
'Taobao Live': ['taobao', 'taobao live', '淘宝'],
'Kuaishou Live': ['kuaishou', '快手'],
'Amazon Live': ['amazon live', 'amazon'],
'Instagram Live': ['instagram live', 'instagram'],
}
OBJECTIVE_RULES = {
'Launch Conversion': ['launch', 'new product', 'debut', 'conversion', 'sell'],
'Inventory Clearance': ['clearance', 'markdown', 'old stock', 'inventory', 'stock'],
'Education / Trust Building': ['education', 'demo', 'tutorial', 'trust'],
'Audience Growth': ['followers', 'reach', 'traffic', 'audience', 'new viewers'],
}
RISK_RULES = {
'Traffic uncertainty': ['traffic', 'viewers', 'reach', 'low traffic'],
'Inventory pressure': ['inventory', 'stock', 'oos', 'out of stock', 'sell out'],
'Offer complexity': ['bundle', 'coupon', 'offer', 'discount stack', 'voucher'],
'Host confidence gap': ['host', 'rookie', 'script', 'confidence', 'training'],
'Technical / moderation load': ['comment', 'chat', 'lag', 'audio', 'mic', 'camera', 'tech'],
'Compliance sensitivity': ['claim', 'compliance', 'policy', 'regulated'],
}
def _normalize_input(user_input: ShowInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['channel', 'objective', 'products', 'offers', 'team', 'risks', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(item) for item in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _match_one(text: str, rules: Dict[str, Sequence[str]], default: str) -> str:
for label, keywords in rules.items():
if any(keyword in text for keyword in keywords):
return label
return default
def _match_many(text: str, rules: Dict[str, Sequence[str]], default: List[str], limit: int = 4) -> List[str]:
found = [label for label, keywords in rules.items() if any(keyword in text for keyword in keywords)]
ordered: List[str] = []
for item in found + default:
if item not in ordered:
ordered.append(item)
return ordered[:limit]
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'None explicitly provided'
class LiveCommerceShowrunner:
def __init__(self, user_input: ShowInput):
self.raw = user_input
self.text = _normalize_input(user_input)
self.lower = self.text.lower()
self.show_type = _match_one(self.lower, SHOW_TYPE_RULES, 'Product Launch Live')
self.channels = _match_many(self.lower, CHANNEL_RULES, ['Douyin Live', 'TikTok Shop Live'], limit=4)
self.objective = _match_one(self.lower, OBJECTIVE_RULES, 'Launch Conversion')
self.risks = _match_many(self.lower, RISK_RULES, ['Traffic uncertainty', 'Technical / moderation load'], limit=4)
def _run_of_show(self) -> List[List[str]]:
opening = 'Open with one clear reason to stay, one hero product angle, and one timed incentive.'
if self.show_type == 'Education / Demo Session':
opening = 'Open with the problem statement, who the product is for, and the learning promise for the session.'
elif self.show_type == 'Promo / Clearance Push':
opening = 'Open with urgency, inventory reality, and the first easy-to-understand deal.'
middle = 'Demonstrate proof, compare options, answer objections, then stack offers without confusing the viewer.'
if self.show_type == 'Guest / Co-host Session':
middle = 'Use the guest for social proof or expertise, then hand back to the host for the sell-through and CTA.'
closing = 'Summarize the hero offer, restate urgency, and direct viewers to the checkout path before ending.'
return [
['0-3 min', 'Hook and room setup', opening],
['3-12 min', 'Proof and demonstration', middle],
['12-22 min', 'Offer stack and objection handling', 'Rotate between proof, price logic, coupon clarity, and live Q&A.'],
['22-30 min', 'Conversion push', 'Bring back hero SKUs, reinforce scarcity, and isolate one checkout CTA at a time.'],
['30 min+', 'Close or extension', closing],
]
def _offer_notes(self) -> List[str]:
notes = [
'Limit the number of simultaneous incentives so viewers can understand the buy path quickly.',
'Pair each hero SKU with one proof point, one price anchor, and one action cue.',
'Keep one rescue offer in reserve in case traffic or conversion starts soft.',
]
if self.objective == 'Inventory Clearance':
notes[1] = 'Use bundle or clear-out logic that moves stock decisively without turning the offer explanation into chaos.'
if 'Compliance sensitivity' in self.risks:
notes.append('Remove any benefit claim or urgency line that cannot be defended by policy or product fact.')
return notes
def render(self) -> str:
lines: List[str] = []
lines.append('# Live Commerce Showrunner Brief')
lines.append('')
lines.append(f'**Show type:** {self.show_type}')
lines.append(f'**Channels in scope:** {_join(self.channels)}')
lines.append(f'**Primary objective:** {self.objective}')
lines.append(f'**Risk flags:** {_join(self.risks)}')
lines.append('**Method note:** This is a heuristic show plan. No live traffic, GMV, comment, or inventory system was accessed.')
lines.append('')
lines.append('## Show Strategy Summary')
lines.append('- Build the session around one commercial story, not a pile of disconnected product moments.')
lines.append('- Keep the host, moderator, and operator aligned on what counts as the hero SKU, rescue offer, and escalation trigger.')
lines.append(f'- Because the main objective is **{self.objective.lower()}**, the show should simplify the next buyer action instead of maximizing noise.')
lines.append('')
lines.append('## Run of Show')
lines.append('| Segment | Focus | Operator note |')
lines.append('|---|---|---|')
for segment, focus, note in self._run_of_show():
lines.append(f'| {segment} | {focus} | {note} |')
lines.append('')
lines.append('## Offer and Merch Plan')
for item in self._offer_notes():
lines.append(f'- {item}')
lines.append('')
lines.append('## Host and Crew Checklist')
lines.append('- **Host:** own the story arc, product proof, CTA clarity, and energy pacing.')
lines.append('- **Moderator:** surface real buyer objections, repeat the checkout path, and keep comments usable.')
lines.append('- **Operator:** track timing, price or coupon readiness, and risk triggers like lag or stock pressure.')
lines.append('- **Product support:** prepare proof assets, comparison points, and a fast answer bank for likely objections.')
lines.append('')
lines.append('## Comment Moderation and Conversion Plays')
lines.append('- Answer repeat objections with short, consistent replies so the host does not need to restart the explanation every minute.')
lines.append('- Pin the current CTA and replace it only when the offer or hero SKU changes.')
lines.append('- Escalate payment, stock, or policy confusion immediately instead of improvising inaccurate answers on stream.')
lines.append('')
lines.append('## Failure Modes and Backup Plan')
lines.append('- If traffic starts soft, shorten the intro, bring forward proof, and trigger one easy-to-understand offer earlier.')
lines.append('- If comments get chaotic, reduce the number of offers on screen and return to one SKU plus one CTA.')
lines.append('- If a product sells out or becomes risky, pivot to the backup SKU and explicitly reset the viewer expectation.')
lines.append('- If audio, lag, or moderation breaks, the operator should have a stop-or-reset threshold instead of hoping the issue disappears.')
lines.append('')
lines.append('## Metrics and Debrief')
lines.append('- Review traffic quality, add-to-cart or click intent, objection frequency, and where attention dropped.')
lines.append('- Compare the planned offer ladder to what actually created response or confusion.')
lines.append('- Note the three moments to keep, change, and remove before the next live session.')
lines.append('')
lines.append('## Assumptions and Limits')
lines.append('- This plan is only as strong as the offer, product, and host context provided by the user.')
lines.append('- Conversion and GMV outcomes depend on traffic quality, compliance, inventory, execution, and platform conditions.')
lines.append('- Final pricing, legal, moderation, and inventory decisions remain human-approved.')
return "\n".join(lines)
def handle(user_input: ShowInput) -> str:
return LiveCommerceShowrunner(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import LiveCommerceShowrunner, handle
def test_channel_detection_douyin_tiktok():
show = LiveCommerceShowrunner('Plan a Douyin and TikTok Shop launch live for our new product.')
assert 'Douyin Live' in show.channels
assert 'TikTok Shop Live' in show.channels
def test_show_type_detection_promo():
show = LiveCommerceShowrunner('We need a clearance livestream with discount bundles and markdown pressure.')
assert show.show_type == 'Promo / Clearance Push'
def test_risk_detection_inventory_and_tech():
show = LiveCommerceShowrunner('Traffic may be low, inventory is tight, and we are worried about audio lag in comments.')
assert 'Traffic uncertainty' in show.risks
assert 'Inventory pressure' in show.risks
assert 'Technical / moderation load' in show.risks
def test_render_contains_sections():
output = handle('Create a run of show for a 45-minute Douyin live with a host and moderator.')
assert output.startswith('# Live Commerce Showrunner Brief')
assert '## Run of Show' in output
assert '## Host and Crew Checklist' in output
assert '## Failure Modes and Backup Plan' in output
def test_dict_input_supported():
output = handle({
'channel': ['TikTok Shop', 'Douyin'],
'objective': 'launch conversion',
'products': ['serum', 'bundle'],
'risks': ['inventory tight', 'tech rehearsal needed'],
})
assert '# Live Commerce Showrunner Brief' in output
assert 'Launch Conversion' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Synthesize ecommerce operational health across channels, inventory, marketing efficiency, customer metrics, and fulfillment into a single operator pulse brie...
---
name: ecommerce-pulse-board
description: Synthesize ecommerce operational health across channels, inventory, marketing efficiency, customer metrics, and fulfillment into a single operator pulse brief. Use when founders, ops leads, or channel managers need a periodic health-check summary that surfaces what is working, what is at risk, and what needs immediate attention — without live BI, ERP, or dashboard API integrations.
---
# Ecommerce Pulse Board
## Overview
Use this skill to generate a structured operational pulse brief from rough KPI notes, multi-channel context, and operational observations. It applies a built-in ecommerce health framework across five pillars — traffic, conversion, inventory, marketing efficiency, and fulfillment — to surface signal, risk, and priority action items in one readable digest.
This MVP is heuristic. It does **not** connect to live analytics platforms, ad dashboards, ERPs, or order management systems. It relies on the user's provided KPI notes, channel mix, and operational context.
## Trigger
Use this skill when the user wants a:
- daily or weekly operational pulse check across all ecommerce channels
- monthly executive health brief for a brand or store
- post-campaign operational debrief covering inventory, fulfillment, and margin side effects
- rapid health scan before a board meeting, investor update, or team sync
- structured digest when data is partial or non-standard across channels
### Example prompts
- "Give me a pulse check on our Shopify and Amazon operations this week"
- "Create a monthly ecommerce health brief from these KPI notes"
- "Run a quick health scan before our investor update"
- "What should our ops team be paying attention to this week?"
## Workflow
1. Capture the review window, channels in scope, and the user's main concern or question.
2. Organize KPI notes and operational observations into the five health pillars.
3. Score each pillar (green / yellow / red) with supporting evidence.
4. Surface the top 3 priority signals and recommended owner actions.
5. Return a markdown pulse brief.
## Inputs
The user can provide any mix of:
- review window: daily, weekly, monthly, post-campaign, quarterly
- channel mix: Shopify, Amazon, TikTok Shop, JD, Taobao, WeChat, Weidian, other
- KPI notes: GMV, net revenue, orders, AOV, refund rate, ROAS, CPC, inventory levels, stockout SKUs
- operational context: staffing, warehouse delays, campaign context, product launch
- main concern or question: e.g., "Are we on track for Q2 targets?" or "Did the promo hurt margin?"
## Outputs
Return a markdown pulse brief with:
- pulse summary: overall health score, review window, and main story in 3 sentences
- pillar health grid: traffic, conversion, inventory, marketing efficiency, fulfillment — each scored and annotated
- signal of the week: the single most important observation and what it means
- risk watchlist: top 2-3 at-risk areas with likely cause and recommended owner
- priority actions: 3-5 specific, operator-ready next steps with owner assignment
- channel spotlight: per-channel pulse for each channel in scope
- data quality note: what was confirmed, what was inferred, and what needs verification
## Safety
- No live analytics, ad platform, ERP, or OMS access.
- Health scores are directional unless comprehensive data is provided.
- Do not fabricate KPI values or claim live data access.
- Strategic budget, pricing, and inventory decisions remain human-approved.
## Best-fit Scenarios
- founders and ops leads who need a quick weekly or monthly health check without dashboard tooling
- teams with fragmented data across platforms that need a single structured digest
- post-campaign or seasonal debriefs that need an operational lens beyond pure media metrics
## Not Ideal For
- real-time operational monitoring, automated alerting, or live KPI dashboards
- enterprises requiring formal financial reporting or audit-grade data reconciliation
- situations where data is so incomplete that any health score would be misleading
## Acceptance Criteria
- Return markdown text.
- Include pillar health grid, signal of the week, risk watchlist, and priority actions.
- Cover at least 3 of the 5 health pillars.
- Make data-quality assumptions explicit when inputs are partial.
- Keep the brief practical for founders, ops leads, and channel managers.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
REVIEW_WINDOW_RULES = {
'Daily': ['daily', 'today', 'yesterday', 'morning', 'day'],
'Weekly': ['weekly', 'this week', 'last week', 'week'],
'Monthly': ['monthly', 'this month', 'last month', 'month'],
'Post-Campaign': ['post-campaign', 'after campaign', 'campaign debrief'],
'Quarterly': ['quarterly', 'quarter', 'q1', 'q2', 'q3', 'q4'],
}
CHANNEL_RULES = {
'Shopify': ['shopify', 'dtc', 'storefront'],
'Amazon': ['amazon'],
'TikTok Shop': ['tiktok', 'douyin'],
'JD': ['jd', 'jingdong'],
'Taobao': ['taobao', '1688'],
'WeChat / Weidian': ['wechat', 'weixin', 'weidian'],
'Other': ['other', 'general'],
}
PILLAR_RULES = {
'Traffic': ['traffic', 'visit', 'session', 'click', 'impression', 'reach', 'ctr'],
'Conversion': ['conversion', 'cvr', 'checkout', 'cart', 'atc', 'order', 'purchase'],
'Inventory': ['inventory', 'stock', 'stockout', 'oos', 'backorder', 'cover', 'aging'],
'Marketing Efficiency': ['roas', 'mer', 'cpm', 'cpc', 'cpa', 'ad spend', 'spend'],
'Fulfillment': ['fulfillment', 'shipping', 'delivery', 'od', 'late shipment', 'oms', 'warehouse'],
}
CONCERN_RULES = {
'GMV / Revenue': ['gmv', 'revenue', 'sales', 'orders'],
'Margin / Profitability': ['margin', 'profit', 'gross', 'refund'],
'Traffic / Channel': ['traffic', 'channel', 'traffic drop'],
'Inventory / Stockout': ['stockout', 'inventory', 'stock', 'cover'],
'Ad Efficiency': ['roas', 'mer', 'ad spend', 'efficiency'],
'Fulfillment': ['shipping', 'delivery', 'fulfillment', 'oms'],
}
PulseInput = Union[str, Dict[str, Any]]
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for kw in kws if kw in text) for name, kws in rules.items()}
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'Not specified'
class EcommercePulseBoard:
def __init__(self, user_input: PulseInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.window = self._detect_window()
self.channels = self._detect_channels()
self.pillars = self._detect_pillars()
self.concern = self._detect_concern()
def _normalize_input(self, user_input: PulseInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['window', 'channels', 'kpis', 'concern', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(v) for v in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_window(self) -> str:
scores = _score_rules(self.lower, REVIEW_WINDOW_RULES)
best = max(scores, key=scores.get)
return best if scores[best] > 0 else 'Weekly'
def _detect_channels(self) -> List[str]:
matched = [name for name, kws in CHANNEL_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Shopify', 'Amazon']
def _detect_pillars(self) -> List[str]:
matched = [name for name, kws in PILLAR_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Traffic', 'Conversion', 'Marketing Efficiency']
def _detect_concern(self) -> str:
scores = _score_rules(self.lower, CONCERN_RULES)
if scores and max(scores.values()) > 0:
return max(scores, key=scores.get)
return 'Overall Health'
def _pillar_scores(self) -> List[Dict[str, str]]:
pillar_notes = {
'Traffic': {
'signal': 'Sessions, clicks, reach; direction vs. prior period.',
'check': 'Which channel drove the most traffic? Any sudden channel shifts?',
'owner': 'Paid Media + Organic',
},
'Conversion': {
'signal': 'CVR, checkout completion, cart abandonment rate.',
'check': 'Did CVR hold despite traffic changes? Any funnel step spikes?',
'owner': 'Storefront + Product',
},
'Inventory': {
'signal': 'Days of cover, stockout count, inbound ETA confidence.',
'check': 'Any hero SKU at risk of stockout before next inbound?',
'owner': 'Ops + Purchasing',
},
'Marketing Efficiency': {
'signal': 'ROAS, MER, CPM, CPC; blended vs. channel-level.',
'check': 'Did ROAS hold? Any channel efficiency degradation?',
'owner': 'Paid Media',
},
'Fulfillment': {
'signal': 'OTIF %, late shipment rate, average delivery time.',
'check': 'Any warehouse or carrier capacity constraints?',
'owner': 'Ops + Logistics',
},
}
scores = {
'Traffic': '🟡 Moderate',
'Conversion': '🟡 Moderate',
'Inventory': '🟡 Moderate',
'Marketing Efficiency': '🟡 Moderate',
'Fulfillment': '🟢 Good',
}
if self.concern == 'Traffic / Channel':
scores['Traffic'] = '🔴 Weak'
if self.concern == 'GMV / Revenue':
scores['Conversion'] = '🔴 Weak'
if self.concern == 'Ad Efficiency':
scores['Marketing Efficiency'] = '🔴 Weak'
if self.concern == 'Inventory / Stockout':
scores['Inventory'] = '🔴 Weak'
if self.concern == 'Fulfillment':
scores['Fulfillment'] = '🔴 Weak'
rows = []
for pillar in ['Traffic', 'Conversion', 'Inventory', 'Marketing Efficiency', 'Fulfillment']:
if pillar in self.pillars:
info = pillar_notes[pillar]
rows.append({
'pillar': pillar,
'score': scores.get(pillar, '🟡 Moderate'),
'signal': info['signal'],
'check': info['check'],
'owner': info['owner'],
})
return rows
def _signal_of_week(self) -> str:
concern_signals = {
'GMV / Revenue': 'Top-line demand direction this week; whether volume is up, flat, or down and which channel is driving the change.',
'Margin / Profitability': 'Refund rate and promotional discount depth; whether margin is being protected as volume changes.',
'Traffic / Channel': 'Channel mix and whether traffic sources are shifting in quality or cost.',
'Inventory / Stockout': 'Hero SKU cover and whether any inbound risk is emerging before the next campaign.',
'Ad Efficiency': 'ROAS trend vs. prior 7 days; whether any channel is becoming inefficient.',
'Fulfillment': 'OTIF and late shipment rate; any carrier or warehouse pressure building up.',
'Overall Health': 'The single most material observation from the week across channels and pillars.',
}
return concern_signals.get(self.concern, concern_signals['Overall Health'])
def _risk_watchlist(self) -> List[Dict[str, str]]:
risks = [
{
'area': 'Traffic Concentration',
'cause': 'Over-reliance on a single paid channel creates vulnerability if ROAS declines.',
'owner': 'Paid Media',
'mitigation': 'Diversify to at least 2 channels; build organic or email as a hedge.',
},
{
'area': 'Inventory Risk at Hero SKU',
'cause': 'Top SKUs with low days of cover and uncertain inbound timing.',
'owner': 'Ops + Purchasing',
'mitigation': 'Lock reorder for hero SKUs with <14 days cover; do not wait for sell-through confirmation.',
},
{
'area': 'Conversion Funnel Weakness',
'cause': 'Cart abandonment or checkout drop-off that is not being actively diagnosed.',
'owner': 'Storefront',
'mitigation': 'Set up weekly funnel review; address the highest-drop-off step first.',
},
]
if self.concern == 'Ad Efficiency':
risks.insert(0, {
'area': 'ROAS Decline in Paid Channel',
'cause': 'Creative fatigue, audience exhaustion, or landing page conversion drop.',
'owner': 'Paid Media',
'mitigation': 'Pause declining campaign, test new creative, and audit landing page before scaling back.',
})
return risks[:3]
def _priority_actions(self) -> List[str]:
actions = [
'Review the signal of the week and confirm whether it is a trend or a one-time event.',
'Check inventory cover for the top 3 SKUs — lock reorder if days of cover is below 14.',
'Run a 7-day ROAS check across all paid channels; flag any channel down >15% vs. prior week.',
'Review the top 3 conversion drop-off steps in the checkout funnel.',
'Confirm fulfillment OTIF for the week; escalate any carrier delays before they compound.',
]
if self.concern == 'Traffic / Channel':
actions.insert(0, 'Audit traffic source mix: paid vs. organic vs. referral. Identify the channel with the most leverage.')
if self.concern == 'Inventory / Stockout':
actions.insert(0, 'Pull a stockout report for all SKUs with <10 days of cover. Assign an owner and ETA for each.')
return actions[:5]
def _channel_spotlight(self) -> List[str]:
lines = []
for ch in self.channels:
lines.append(f'### {ch}')
lines.append(f'- Track: orders, AOV, refund rate, and top SKU performance.')
lines.append(f'- Watch: stock availability, campaign performance, and customer review velocity.')
lines.append('')
return lines
def render(self) -> str:
lines: List[str] = []
lines.append('# Ecommerce Pulse Brief')
lines.append('')
lines.append(f'**Review window:** {self.window}')
lines.append(f'**Channels in scope:** {_join(self.channels)}')
lines.append(f'**Main concern:** {self.concern}')
lines.append('**Method note:** This is a heuristic operational brief. No live analytics, ad platform, ERP, or OMS was accessed.')
lines.append('')
lines.append('## Pulse Summary')
lines.append(f'- **{self.window} review** across {_join(self.channels)}.')
lines.append(f'- Primary concern this cycle: **{self.concern}**.')
lines.append('- Assess each pillar below; treat 🟡 as "monitor closely" and 🔴 as "action required today."')
lines.append('')
lines.append('## Pillar Health Grid')
lines.append('| Pillar | Score | Signal | First Check | Owner |')
lines.append('|---|---|---|---|---|')
for row in self._pillar_scores():
lines.append(f'| {row["pillar"]} | {row["score"]} | {row["signal"]} | {row["check"]} | {row["owner"]} |')
lines.append('')
lines.append('## Signal of the Week')
lines.append(f'- **{self.concern}:** {self._signal_of_week()}')
lines.append('')
lines.append('## Risk Watchlist')
lines.append('| At-Risk Area | Likely Cause | Recommended Owner | Mitigation |')
lines.append('|---|---|---|---|---|')
for risk in self._risk_watchlist():
lines.append(f'| {risk["area"]} | {risk["cause"]} | {risk["owner"]} | {risk["mitigation"]} |')
lines.append('')
lines.append('## Priority Actions')
for idx, action in enumerate(self._priority_actions(), 1):
lines.append(f'{idx}. {action}')
lines.append('')
lines.append('## Channel Spotlight')
for item in self._channel_spotlight():
lines.append(item)
lines.append('## Data Quality Note')
lines.append('- KPI values in this brief are directional. Confirm actual numbers against your analytics, ad platform, or ERP before making irreversible decisions.')
lines.append('- If any pillar is missing data, treat it as 🟡 by default until confirmed.')
lines.append('- Cross-reference channel-level data against total business metrics to avoid double-counting or missing attribution gaps.')
return '\n'.join(lines)
def handle(user_input: PulseInput) -> str:
return EcommercePulseBoard(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import EcommercePulseBoard, handle
def test_window_detection_weekly():
pulse = EcommercePulseBoard('Give me a weekly pulse check on our Shopify and Amazon operations')
assert pulse.window == 'Weekly'
def test_window_detection_monthly():
pulse = EcommercePulseBoard('Monthly ecommerce health brief from these KPI notes')
assert pulse.window == 'Monthly'
def test_channel_detection_shopify_amazon():
pulse = EcommercePulseBoard('Pulse check for Shopify and Amazon this week')
assert 'Shopify' in pulse.channels
assert 'Amazon' in pulse.channels
def test_concern_detection_gmv():
pulse = EcommercePulseBoard('GMV dropped this week — what is happening?')
assert pulse.concern == 'GMV / Revenue'
def test_pillar_detection_traffic_conversion():
pulse = EcommercePulseBoard('Weekly review of traffic and conversion metrics')
assert 'Traffic' in pulse.pillars
assert 'Conversion' in pulse.pillars
def test_render_contains_sections():
output = handle('Give me a pulse check on our Shopify and Amazon operations this week')
assert output.startswith('# Ecommerce Pulse Brief')
assert '## Pillar Health Grid' in output
assert '## Signal of the Week' in output
assert '## Risk Watchlist' in output
assert '## Priority Actions' in output
def test_dict_input_supported():
output = handle({
'window': 'Weekly',
'channels': ['Shopify', 'Amazon', 'TikTok Shop'],
'concern': 'Overall Health',
'kpis': {'GMV': 'down 5%', 'ROAS': 'stable'}
})
assert '# Ecommerce Pulse Brief' in output
assert 'Weekly' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Define CRM segments (new, active, lapsed, VIP, at-risk, churned), build winback triggers, and generate personalized re-engagement campaign ideas across email...
---
name: crm-segment-winback
description: Define CRM segments (new, active, lapsed, VIP, at-risk, churned), build winback triggers, and generate personalized re-engagement campaign ideas across email, SMS, WeChat, and app push. Use when CRM managers, operators, or founders need a segment health brief, winback strategy, or lifecycle campaign calendar without live CRM API, marketing automation, or analytics platform access.
---
# CRM Segment & Winback
## Overview
Use this skill to turn rough customer data notes and lifecycle goals into a structured CRM segment brief and winback campaign strategy. It applies a built-in lifecycle stage framework, winback trigger library, and channel suitability matrix to generate operator-ready recommendations.
This MVP is heuristic. It does **not** connect to live CRM platforms (Klaviyo, Braze, SMS tools, WeChat Work), analytics, or customer data warehouses. It relies on the user's provided segment context, purchase history notes, and campaign goals.
## Trigger
Use this skill when the user wants to:
- define or refine CRM customer segments (new, active, lapsed, VIP, at-risk, churned)
- build winback triggers and timing for lapsed or churned customers
- plan a lifecycle email or SMS campaign sequence across onboarding, retention, and winback
- assess segment health and identify which segments need immediate re-engagement
- generate personalized offer and message ideas for a specific segment
### Example prompts
- "Help me define our CRM segments for a DTC skincare brand"
- "Build a winback strategy for customers who haven't purchased in 90 days"
- "Create a lifecycle campaign calendar for new customers, VIPs, and lapsed buyers"
- "What SMS and email sequence should we use to reactivate churned subscribers?"
## Workflow
1. Capture the business context, customer lifecycle stage, and campaign goal.
2. Define or refine the segment framework based on recency, frequency, monetary, and engagement signals.
3. Identify winback triggers and optimal re-engagement timing per segment.
4. Map segments and campaign types to channels (email, SMS, WeChat, push).
5. Return a markdown CRM brief with segment definitions, winback triggers, channel matrix, and campaign ideas.
## Inputs
The user can provide any mix of:
- business type: DTC brand, marketplace seller, subscription, hybrid
- existing segment notes: current segment names, definitions, or gaps
- customer data context: purchase frequency, average order value, churn patterns, email/SMS opt-in rates
- campaign goal: reduce churn, increase LTV, reactivate lapsed buyers, promote VIP loyalty
- available channels: email (Klaviyo, Mailchimp), SMS (Klaviyo, SMSBump), WeChat, app push
- lifecycle stage: new customer onboarding, ongoing retention, winback, VIP cultivation
## Outputs
Return a markdown brief with:
- segment framework summary (recency, frequency, monetary, engagement tiers)
- segment health scorecard (size, AOV, churn risk, engagement quality per segment)
- winback trigger library (timing, signal, offer type)
- channel suitability matrix (which channel works best for each segment and campaign type)
- personalized message and offer ideas per segment and channel
- campaign sequence outline (email/SMS flow with subject line ideas and timing)
- KPI framework for segment and campaign performance (open rate, CTR, conversion, LTV)
## Safety
- No live CRM platform, customer data warehouse, or marketing automation access.
- Segment definitions are directional; actual segment assignment depends on data quality and platform logic.
- Do not claim guaranteed re-engagement rates or LTV improvements.
- Offer decisions, discount levels, and frequency management remain human-approved.
## Best-fit Scenarios
- DTC brands and small-to-mid-market ecommerce teams with email and SMS programs
- operators building CRM strategy from scratch or refining an existing fragmented approach
- teams that need winback campaign ideas without deep CRM tool expertise
## Not Ideal For
- real-time customer data analysis, automated segmentation, or live campaign execution
- highly regulated industries with strict privacy or consent requirements (healthcare, financial)
- enterprise-grade CRM setups requiring complex behavioral or predictive modeling
## Acceptance Criteria
- Return markdown text.
- Include segment framework, winback trigger library, and channel matrix.
- Cover at least 3 distinct segments with campaign ideas.
- Make data-quality assumptions explicit when segment notes are partial.
- Keep the brief practical for CRM managers and ecommerce operators.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
BUSINESS_RULES = {
'DTC Brand': ['dtc', 'direct to consumer', 'brand', 'shopify', 'owned'],
'Marketplace Seller': ['amazon', 'jd', 'taobao', 'marketplace', 'platform'],
'Subscription': ['subscription', 'subscribe', 'recurring', 'membership'],
'Hybrid': ['hybrid', 'multi-channel', ' DTC and marketplace'],
}
SEGMENT_RULES = {
'New Customer': ['new', 'first order', 'first purchase', 'onboarding'],
'Active Customer': ['active', 'regular', 'repeat', 'frequent'],
'Lapsed Customer': ['lapsed', 'inactive', 'dormant', 'has not purchased', '90 days'],
'VIP / Loyal': ['vip', 'loyal', 'top', 'best', 'high value', 'lifetime'],
'At-Risk': ['at risk', 'at-risk', 'declining', 'decreasing', 'reduced'],
'Churned': ['churned', 'churn', 'lost', 'unsubscribed', 'opted out'],
}
CHANNEL_RULES = {
'Email': ['email', 'klaviyo', 'mailchimp', 'newsletter'],
'SMS': ['sms', 'text', 'message', 'smsbump'],
'WeChat': ['wechat', 'weixin', 'miniprogram'],
'App Push': ['push', 'app notification', 'mobile push'],
}
WinbackInput = Union[str, Dict[str, Any]]
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for kw in kws if kw in text) for name, kws in rules.items()}
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'Not specified'
class CRMSegmentWinback:
def __init__(self, user_input: WinbackInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.business = self._detect_business()
self.segments = self._detect_segments()
self.channels = self._detect_channels()
self.goal = self._detect_goal()
def _normalize_input(self, user_input: WinbackInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['business', 'segments', 'channels', 'campaign_goal', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(v) for v in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_business(self) -> str:
scores = _score_rules(self.lower, BUSINESS_RULES)
best = max(scores, key=scores.get)
return best if scores[best] > 0 else 'DTC Brand'
def _detect_segments(self) -> List[str]:
matched = [name for name, kws in SEGMENT_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['New Customer', 'Active Customer', 'Lapsed Customer']
def _detect_channels(self) -> List[str]:
matched = [name for name, kws in CHANNEL_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Email']
def _detect_goal(self) -> str:
if any(kw in self.lower for kw in ['winback', 'reengage', 're-activate', 'reactivate', 'lapsed', 'churn']):
return 'Winback / Re-engagement'
if any(kw in self.lower for kw in ['vip', 'loyal', 'lifetime', 'retention']):
return 'VIP Loyalty'
if any(kw in self.lower for kw in ['onboarding', 'new customer', 'first order']):
return 'New Customer Onboarding'
return 'Segment Health Review'
def _segment_framework(self) -> List[Dict[str, str]]:
return [
{
'segment': 'New Customer',
'definition': 'First order completed within last 30 days',
'recency': '0–30 days',
'frequency': '1 order',
'monetary': 'First-order AOV',
'engagement': 'Email opens >20%, SMS click >5%',
'priority': 'High',
},
{
'segment': 'Active Customer',
'definition': '2+ orders in last 90 days, AOV above median',
'recency': '0–90 days',
'frequency': '2+ orders',
'monetary': 'Above median AOV',
'engagement': 'Email opens >30%, SMS click >8%',
'priority': 'Maintain',
},
{
'segment': 'Lapsed Customer',
'definition': 'No purchase in 60–180 days, previously active',
'recency': '60–180 days',
'frequency': 'Was 2+, now 0',
'monetary': 'Below or at median AOV',
'engagement': 'Email opens <15%, disengaged',
'priority': 'High — winback window',
},
{
'segment': 'At-Risk',
'definition': 'Declining purchase frequency or AOV in last 60 days',
'recency': '0–60 days but declining',
'frequency': 'Frequency dropping MoM',
'monetary': 'AOV declining',
'engagement': 'Email opens <20%, SMS沉默',
'priority': 'Urgent — early intervention',
},
{
'segment': 'VIP / Loyal',
'definition': 'Top 10% by LTV, 3+ orders in last 6 months, no refund issues',
'recency': '0–60 days',
'frequency': '3+ orders',
'monetary': 'Top 10% LTV',
'engagement': 'High across all channels',
'priority': 'Protect — exclusive offers',
},
{
'segment': 'Churned',
'definition': 'No purchase in 180+ days, email/SMS also disengaged',
'recency': '180+ days',
'frequency': 'Previously active, now 0',
'monetary': 'Variable, often below median',
'engagement': 'Unsubscribed or fully disengaged',
'priority': 'Low priority — expensive to win back',
},
]
def _winback_triggers(self) -> List[Dict[str, str]]:
return [
{'trigger': 'No purchase in 30 days', 'segment': 'Active → Lapsed', 'action': 'Send a re-engagement email with best-seller recommendation.'},
{'trigger': 'No purchase in 60 days', 'segment': 'Lapsed', 'action': 'SMS winback with 10–15% off incentive; email follow-up with social proof.'},
{'trigger': 'No purchase in 90 days', 'segment': 'Lapsed → At-Risk', 'action': 'Personalized "we miss you" email + exclusive offer + UGC reminder.'},
{'trigger': 'Frequency declining (MoM)', 'segment': 'Active → At-Risk', 'action': 'Proactive outreach; offer a loyalty reward or bundle to stabilize.'},
{'trigger': 'Cart abandoned (email)', 'segment': 'Any active segment', 'action': 'Automated 3-step cart abandonment sequence.'},
{'trigger': '180+ days, unsubscribed', 'segment': 'Churned', 'action': 'Consider suppression; reactivation cost often exceeds LTV.'},
]
def _channel_matrix(self) -> List[str]:
lines = ['| Segment | Best Channel | Message Type | Frequency | Offer Tone |']
lines.append('|---|---|---|---|---|')
matrix = [
('New Customer', 'Email', 'Onboarding sequence + product education', 'Email d1/d3/d7', 'Value-add, no discount'),
('Active Customer', 'Email + SMS', 'New arrival + loyalty points', 'Weekly email, bi-weekly SMS', 'Appreciation + early access'),
('Lapsed Customer', 'SMS + Email', 'Winback offer + social proof', 'SMS d1/d7, Email d3', 'Discount or exclusive offer'),
('At-Risk', 'SMS + App Push', 'Urgency + personalized product', 'SMS d1/d3', 'Small incentive to reactivate'),
('VIP / Loyal', 'Email + WeChat', 'Exclusive preview + VIP-only offer', 'Monthly exclusive', 'Recognition, no heavy discount'),
('Churned', 'Email only', 'Last-chance re-engagement', 'One-time email', 'Deep discount or free gift with order'),
]
for row in matrix:
lines.append(f'| {row[0]} | {row[1]} | {row[2]} | {row[3]} | {row[4]} |')
return lines
def _campaign_sequence(self) -> List[str]:
if self.goal == 'Winback / Re-engagement':
return [
'Day 1: SMS — "We miss you — here\'s 15% off your next order [CODE]"',
'Day 3: Email — "Top sellers this week" + personalized product recommendations',
'Day 7: Email — "A customer like you loved [specific product]" + UGC social proof',
'Day 14: Optional — Second discount email or free shipping offer for first order back',
]
if self.goal == 'New Customer Onboarding':
return [
'Day 1: Welcome email — brand story + what\'s in the package',
'Day 3: Product education email — how to use, tips, FAQs',
'Day 7: Cross-sell email — complementary product based on first purchase',
'Day 14: Review request email — request product review + social share',
'Day 30: Reactivation check — has this customer purchased again? If not, move to lapsed flow.',
]
if self.goal == 'VIP Loyalty':
return [
'Monthly: Exclusive early access to new product launches',
'Quarterly: VIP-only discount or free shipping event',
'Ongoing: Personalized birthday/anniversary offer',
'Post-purchase: Thank-you note + loyalty points bonus',
]
return [
'Map each segment to the appropriate campaign flow above.',
'Set up triggered sends for behavioral signals (abandoned cart, browse, review).',
'Review segment size and engagement monthly; suppress disengaged contacts to protect deliverability.',
]
def _kpi_framework(self) -> List[str]:
return [
'**Segment size:** Count of contacts per segment, updated monthly.',
'**Email open rate:** Target >25% for winback, >35% for VIP; <15% signals disengagement.',
'**SMS click rate:** Target >5%; below 2% suggests message or offer fatigue.',
'**Reactivation rate:** % of lapsed customers who purchase within 30 days of winback campaign.',
'**LTV of reactivated customers:** Compare against new customer LTV to assess winback ROI.',
'**Opt-out rate:** Target <0.5% per campaign; spike indicates message frequency or relevance issue.',
]
def render(self) -> str:
lines: List[str] = []
lines.append('# CRM Segment & Winback Brief')
lines.append('')
lines.append(f'**Business type:** {self.business}')
lines.append(f'**Campaign goal:** {self.goal}')
lines.append(f'**Segments in scope:** {_join(self.segments)}')
lines.append(f'**Channels:** {_join(self.channels)}')
lines.append('**Method note:** This is a heuristic CRM brief. No live CRM platform, customer data warehouse, or marketing automation tool was accessed.')
lines.append('')
lines.append('## Segment Framework')
lines.append('| Segment | Definition | Recency | Frequency | Monetary | Engagement Signal | Priority |')
lines.append('|---|---|---|---|---|---|---|')
for row in self._segment_framework():
lines.append(f'| {row["segment"]} | {row["definition"]} | {row["recency"]} | {row["frequency"]} | {row["monetary"]} | {row["engagement"]} | {row["priority"]} |')
lines.append('')
lines.append('## Winback Trigger Library')
lines.append('| Trigger | Segment Transition | Recommended Action |')
lines.append('|---|---|---|')
for row in self._winback_triggers():
lines.append(f'| {row["trigger"]} | {row["segment"]} | {row["action"]} |')
lines.append('')
lines.append('## Channel Suitability Matrix')
for row in self._channel_matrix():
lines.append(row)
lines.append('')
lines.append('## Campaign Sequence Ideas')
for item in self._campaign_sequence():
lines.append(f'- {item}')
lines.append('')
lines.append('## KPI Framework')
for item in self._kpi_framework():
lines.append(f'- {item}')
lines.append('')
lines.append('## Data Quality Notes')
lines.append('- Confirm RFM (Recency, Frequency, Monetary) definitions are consistent across all platforms before importing into a CRM tool.')
lines.append('- Suppress unsubscribed and permanently bounced contacts from all non-permission-based campaigns.')
lines.append('- Tag each contact with lifecycle stage so the system can auto-transition them over time.')
return '\n'.join(lines)
def handle(user_input: WinbackInput) -> str:
return CRMSegmentWinback(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import CRMSegmentWinback, handle
def test_business_detection_dtc():
crm = CRMSegmentWinback('DTC skincare brand CRM segmentation')
assert crm.business == 'DTC Brand'
def test_segment_detection_lapsed():
crm = CRMSegmentWinback('Winback strategy for customers who have not purchased in 90 days')
assert 'Lapsed Customer' in crm.segments
def test_segment_detection_vip():
crm = CRMSegmentWinback('VIP loyalty program for top-spending customers')
assert 'VIP / Loyal' in crm.segments
def test_channel_detection_email_sms():
crm = CRMSegmentWinback('Email and SMS winback campaign for lapsed customers')
assert 'Email' in crm.channels
assert 'SMS' in crm.channels
def test_goal_detection_winback():
crm = CRMSegmentWinback('Reactivate churned subscribers with email and SMS sequence')
assert crm.goal == 'Winback / Re-engagement'
def test_render_contains_sections():
output = handle('Build a winback strategy for customers who have not purchased in 90 days')
assert output.startswith('# CRM Segment & Winback Brief')
assert '## Segment Framework' in output
assert '## Winback Trigger Library' in output
assert '## Campaign Sequence Ideas' in output
def test_dict_input_supported():
output = handle({
'business': 'DTC Brand',
'segments': ['New Customer', 'Lapsed Customer', 'VIP / Loyal'],
'channels': ['Email', 'SMS'],
'campaign_goal': 'Winback / Re-engagement'
})
assert '# CRM Segment & Winback Brief' in output
assert 'Winback' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Plan, organize, and audit ecommerce content assets across short video (TikTok/Douyin), Xiaohongshu posts, Amazon A+ content, Shopify PDP images, email banner...
---
name: content-asset-orchestrator
description: Plan, organize, and audit ecommerce content assets across short video (TikTok/Douyin), Xiaohongshu posts, Amazon A+ content, Shopify PDP images, email banners, and Meta/Instagram creative. Use when content teams need a content production brief, asset gap audit, workflow calendar, or multi-channel content strategy without live DAM, analytics, or social API access.
---
# Content Asset Orchestrator
## Overview
Use this skill to turn rough content goals, channel priorities, and production constraints into an organized content operations brief. It applies a built-in content type framework, channel requirement matrix, and production workflow to generate an actionable asset plan.
This MVP is heuristic. It does **not** connect to live social accounts, DAM systems, analytics platforms, or content management tools. It relies on the user's provided content context, channel mix, and production resources.
## Trigger
Use this skill when the user wants to:
- plan a content production schedule across TikTok, Xiaohongshu, Amazon, Shopify, email, or Meta
- audit content assets for a product launch, campaign, or seasonal push
- identify content gaps across format types (short video, image, copy, UGC)
- build a content brief for a creator, photographer, or copywriter
- sequence content types across awareness, consideration, and conversion stages
### Example prompts
- "Help me plan content for our summer campaign across TikTok, Xiaohongshu, and Amazon A+"
- "Audit our current content assets for gaps before our product launch"
- "Create a content production brief for our DTC brand's new SKU"
- "What content should we produce this month for our Shopify and Meta channels?"
## Workflow
1. Capture the campaign or production window, target channels, and content goals.
2. Map content types to channels and identify required formats (video, image, copy, UGC).
3. Audit existing assets for completeness, quality, and channel fit.
4. Identify production gaps and assign owners or resource needs.
5. Return a markdown content operations brief with production calendar and asset brief.
## Inputs
The user can provide any mix of:
- campaign window: specific dates, seasonal push, or general monthly planning
- target channels: TikTok, Douyin, Xiaohongshu, Amazon A+, Shopify PDP, email, Meta, Instagram, WeChat
- content goal: awareness, traffic, conversion, UGC seeding, review collection, SEO
- product or SKU context: new launch, best seller, seasonal, clearance
- existing asset inventory: what formats already exist (video, images, copy, UGC)
- production resources: in-house team, agency, creator, photographer, copywriter
- budget for content production
## Outputs
Return a markdown brief with:
- content strategy summary (goal, channels, audience)
- channel requirement matrix (format, dimension, length, caption style)
- content type coverage map (what exists vs. what is needed)
- production gap analysis and asset priority list
- production calendar by week or content type
- brief for each content asset (video, image, copy, UGC) with key messages and format notes
- distribution and posting cadence recommendations
## Safety
- No live social accounts, analytics, DAM, or content management tool access.
- Content recommendations are directional; platform policies and creative quality determine actual performance.
- Do not claim guaranteed engagement or conversion lift from content assets.
- Brand voice and creative decisions remain with the content team.
## Best-fit Scenarios
- small-to-medium content teams managing 3-8 channels without a dedicated content director
- brands launching new products and needing a structured content production sequence
- operators transitioning from single-channel to multi-channel content operations
## Not Ideal For
- real-time content performance tracking, automated posting, or live social media management
- teams with enterprise-scale content operations requiring DAM or MAM integrations
- workflows that require automated approval routing or compliance review
## Acceptance Criteria
- Return markdown text.
- Include channel requirement matrix, gap analysis, and production calendar.
- Cover at least 3 channels with specific format requirements.
- Make resource assumptions explicit.
- Keep the brief practical for content teams and ecommerce operators.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
CHANNEL_RULES = {
'TikTok / Douyin': ['tiktok', 'douyin', 'short video'],
'Xiaohongshu': ['xiaohongshu', 'rednote', 'xhs'],
'Amazon A+': ['amazon a+', 'amazon a+ content'],
'Shopify PDP': ['shopify', 'pdp', 'product page', 'storefront'],
'Meta / Instagram': ['meta', 'facebook', 'instagram', 'fb', 'ig'],
'Email': ['email', 'newsletter', 'klaviyo', 'mailchimp'],
'WeChat': ['wechat', 'weixin'],
'JD / Taobao': ['jd', 'taobao', '天猫'],
}
GOAL_RULES = {
'Awareness': ['awareness', 'brand', 'reach', 'impression'],
'Traffic': ['traffic', 'click', 'website', 'landing page'],
'Conversion': ['conversion', 'sales', 'direct response', 'dra'],
'UGC Seeding': ['ugc', 'user generated', 'creator', 'seeding'],
'Review Collection': ['review', 'testimonial', 'rating'],
'SEO': ['seo', 'search', 'blog', 'content'],
}
FORMAT_RULES = {
'Short Video': ['video', 'tiktok', 'douyin', 'short form', 'reels', 'short video'],
'Image / Carousel': ['image', 'photo', 'carousel', 'gallery', 'static'],
'Copy / Caption': ['copy', 'caption', 'script', 'text', 'blog post'],
'UGC / Creator': ['ugc', 'creator', 'influencer', 'user generated'],
'A+ / Infographic': ['a+', 'infographic', 'comparison', 'chart'],
}
ContentInput = Union[str, Dict[str, Any]]
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for kw in kws if kw in text) for name, kws in rules.items()}
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'Not specified'
class ContentAssetOrchestrator:
def __init__(self, user_input: ContentInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.channels = self._detect_channels()
self.goals = self._detect_goals()
self.formats = self._detect_formats()
self.existing_assets = self._detect_existing()
def _normalize_input(self, user_input: ContentInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['window', 'channels', 'goals', 'product', 'existing', 'resources', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(v) for v in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_channels(self) -> List[str]:
matched = [name for name, kws in CHANNEL_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['TikTok / Douyin', 'Xiaohongshu', 'Shopify PDP']
def _detect_goals(self) -> List[str]:
matched = [name for name, kws in GOAL_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Conversion']
def _detect_formats(self) -> List[str]:
matched = [name for name, kws in FORMAT_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Short Video', 'Image / Carousel', 'Copy / Caption']
def _detect_existing(self) -> List[str]:
existing = []
if any(kw in self.lower for kw in ['已有', 'existing', 'already have', 'current']):
existing = ['Assets noted as partial — assume most formats need refresh or supplementation.']
return existing
def _channel_matrix(self) -> List[str]:
lines = ['| Channel | Primary Format | Aspect Ratio / Spec | Copy Style | Cadence |']
lines.append('|---|---|---|---|---|')
specs = {
'TikTok / Douyin': ('9:16 Short Video', '9:16, 1080×1920', 'Conversational, hook-first, 3–5s to value', '3–7x/week'),
'Xiaohongshu': ('Image or 9:16 Video', '3:4 or 9:16, 1080×1350', 'Lifestyle, personal, specific use-case', '3–5x/week'),
'Amazon A+': ('A+ Infographic + Image', '1200×600 (module)', 'Feature-benefit, comparison', 'Set once, refresh quarterly'),
'Shopify PDP': ('Hero Image + Gallery', '1200×1200 (square)', 'Benefit-led, SEO-friendly', 'Set per SKU, refresh with new launches'),
'Meta / Instagram': ('Feed Image or Reels', '1:1 or 4:5', 'Brand voice, varied hooks', '1–2x/day'),
'Email': ('Banner + Copy', '600×200 (banner)', 'Direct, benefit-first, CTA-clear', '1–2x/week'),
'WeChat': ('Image + Mini Program', '2:3 or 9:16', 'Formal, relationship-first', '2–4x/week'),
'JD / Taobao': ('Main Image + Video', '800×800 or 9:16', 'Feature-dense, comparison', 'Set per SKU'),
}
for ch in self.channels:
row = specs.get(ch, ('TBD', 'TBD', 'TBD', 'TBD'))
lines.append(f'| {ch} | {row[0]} | {row[1]} | {row[2]} | {row[3]} |')
return lines
def _gap_analysis(self) -> List[str]:
gaps = []
if 'Short Video' in self.formats and 'TikTok / Douyin' in self.channels:
gaps.append('TikTok short video: need 3–5 video concepts with hook, body, and CTA per concept.')
if 'Xiaohongshu' in self.channels:
gaps.append('Xiaohongshu: need 3–5 posts with lifestyle narrative, product mention, and hashtag strategy.')
if 'Amazon A+' in self.channels:
gaps.append('Amazon A+: need infographic modules, comparison chart, and brand story section.')
if 'UGC / Creator' in self.formats or 'UGC Seeding' in self.goals:
gaps.append('UGC: brief for 2–3 creators with talking points, format requirements, and disclosure guidelines.')
if 'Email' in self.channels:
gaps.append('Email: banner image brief + subject line variants + CTA copy.')
if not gaps:
gaps.append('No critical gaps identified from input; proceed with asset production list.')
return gaps
def _priority_list(self) -> List[str]:
priorities = []
if 'Conversion' in self.goals:
priorities.append('1. Hero content (video or image) for the highest-traffic SKU or campaign.')
priorities.append('2. A+ or PDP upgrade if Amazon or Shopify is in scope.')
if 'UGC Seeding' in self.goals:
priorities.append('1. Creator brief and content guidelines before production begins.')
if 'Awareness' in self.goals:
priorities.insert(0, '1. TikTok/Xiaohongshu content pipeline for top-of-funnel reach.')
priorities.append('N. Refresh existing assets that are older than 3 months or underperforming.')
return priorities
def _production_brief(self) -> List[Dict[str, str]]:
briefs = []
for ch in self.channels[:3]:
brief = {'channel': ch, 'asset_type': 'TBD', 'key_message': 'TBD', 'format_notes': 'TBD'}
if ch == 'TikTok / Douyin':
brief['asset_type'] = 'Short Video (3–5 variants)'
brief['key_message'] = 'Hook in first 3 seconds; product benefit or transformation moment.'
brief['format_notes'] = '9:16, 1080×1920, no watermark, subtitle file recommended.'
elif ch == 'Xiaohongshu':
brief['asset_type'] = 'Image Post or Short Video'
brief['key_message'] = 'Real-use scenario, personal experience, specific result or tip.'
brief['format_notes'] = '3:4 image or 9:16 video; first image is the hook.'
elif ch == 'Amazon A+':
brief['asset_type'] = 'A+ Infographic Modules'
brief['key_message'] = 'Feature, benefit, and proof point for each module.'
brief['format_notes'] = '1200×600 per module; brand story + comparison grid recommended.'
elif ch == 'Shopify PDP':
brief['asset_type'] = 'Hero Image + Gallery'
brief['key_message'] = 'Primary benefit above the fold; lifestyle + spec images below.'
brief['format_notes'] = '1200×1200 minimum; WebP format recommended.'
briefs.append(brief)
return briefs
def render(self) -> str:
lines: List[str] = []
lines.append('# Content Asset Orchestrator Brief')
lines.append('')
lines.append(f'**Campaign window:** {self._campaign_window()}')
lines.append(f'**Channels:** {_join(self.channels)}')
lines.append(f'**Content goals:** {_join(self.goals)}')
lines.append(f'**Format priorities:** {_join(self.formats)}')
lines.append('**Method note:** This is a heuristic content planning brief. No live social, DAM, or analytics platform was accessed.')
lines.append('')
lines.append('## Content Strategy Summary')
lines.append(f'- Goal: drive {", ".join(self.goals).lower()} across {", ".join(self.channels[:3])}.')
lines.append('- Lead with high-impact visual content; support with targeted copy and UGC seeding.')
lines.append('- Build a content pipeline that can run 3–5 weeks before major campaign activation.')
lines.append('')
lines.append('## Channel Requirement Matrix')
for row in self._channel_matrix():
lines.append(row)
lines.append('')
lines.append('## Content Type Coverage Map')
lines.append('| Format | In Scope | Existing? | Gap? | Priority |')
lines.append('|---|---|---|---|---|')
for fmt in self.formats:
in_scope = '✅' if fmt in self.formats else '❌'
gap = '⚠️ Gap identified' if any(g.lower().startswith(fmt.lower().split()[0]) for g in self._gap_analysis()) else '✅ OK'
pri = '🔴 High' if fmt in ['Short Video', 'Image / Carousel'] else '🟡 Medium'
lines.append(f'| {fmt} | {in_scope} | Partial assumed | {gap} | {pri} |')
lines.append('')
lines.append('## Production Gap Analysis')
for gap in self._gap_analysis():
lines.append(f'- {gap}')
lines.append('')
lines.append('## Asset Priority List')
for item in self._priority_list():
lines.append(f'- {item}')
lines.append('')
lines.append('## Production Briefs')
for brief in self._production_brief():
lines.append(f'### {brief["channel"]}')
lines.append(f'- **Asset type:** {brief["asset_type"]}')
lines.append(f'- **Key message:** {brief["key_message"]}')
lines.append(f'- **Format notes:** {brief["format_notes"]}')
lines.append('')
lines.append('## Distribution & Posting Cadence')
lines.append('- TikTok: post 3–7x/week; go live with creator at least 1x/week during campaign window.')
lines.append('- Xiaohongshu: post 3–5x/week; seed with creator UGC 2 weeks before conversion push.')
lines.append('- Amazon A+: load 1 week before campaign launch; refresh every 3 months.')
lines.append('- Email: send 1–2x/week to engaged segment; pause for 1 week pre/post major campaign.')
return '\n'.join(lines)
def _campaign_window(self) -> str:
if any(kw in self.lower for kw in ['monthly', 'this month', 'q1', 'q2', 'q3', 'q4']):
return 'This quarter'
if any(kw in self.lower for kw in ['launch', 'new product', 'new sku']):
return 'Launch window (T-4 weeks pre-launch)'
return 'General planning window'
def handle(user_input: ContentInput) -> str:
return ContentAssetOrchestrator(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import ContentAssetOrchestrator, handle
def test_channel_detection_tiktok_xiaohongshu():
orchestrator = ContentAssetOrchestrator('Plan content for TikTok and Xiaohongshu summer campaign')
assert 'TikTok / Douyin' in orchestrator.channels
assert 'Xiaohongshu' in orchestrator.channels
def test_channel_detection_amazon_email():
orchestrator = ContentAssetOrchestrator('Amazon A+ content and email newsletter planning')
assert 'Amazon A+' in orchestrator.channels
assert 'Email' in orchestrator.channels
def test_goal_detection_awareness():
orchestrator = ContentAssetOrchestrator('Awareness campaign content for new brand launch')
assert 'Awareness' in orchestrator.goals
def test_format_detection_video():
orchestrator = ContentAssetOrchestrator('Short video content for TikTok and Douyin')
assert 'Short Video' in orchestrator.formats
def test_render_contains_sections():
output = handle('Help me plan content for our summer campaign across TikTok, Xiaohongshu, and Amazon A+')
assert output.startswith('# Content Asset Orchestrator Brief')
assert '## Channel Requirement Matrix' in output
assert '## Production Gap Analysis' in output
assert '## Production Briefs' in output
def test_dict_input_supported():
output = handle({
'window': 'Q2',
'channels': ['TikTok / Douyin', 'Xiaohongshu', 'Shopify PDP'],
'goals': ['Conversion', 'UGC Seeding'],
'product': 'skincare brand launch'
})
assert '# Content Asset Orchestrator Brief' in output
assert 'TikTok' in output or 'TikTok' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')
Analyze ecommerce ad spend notes across Meta Ads, Google Ads, TikTok Ads, Amazon Sponsored, and Xiaohongshu promotional feeds, then recommend budget realloca...
---
name: ad-budget-rebalancer
description: Analyze ecommerce ad spend notes across Meta Ads, Google Ads, TikTok Ads, Amazon Sponsored, and Xiaohongshu promotional feeds, then recommend budget reallocation across channels, campaign types, and audience segments. Use when operators need a budget review, underperforming-channel diagnosis, or channel-mix rebalancing brief without live ad platform API access.
---
# Ad Budget Rebalancer
## Overview
Use this skill to diagnose ad spend patterns and generate a budget-rebalancing brief that prioritizes channels, campaign types, and audience segments based on efficiency signals. It applies a built-in efficiency framework and channel-mix matrix to surface reallocation recommendations.
This MVP is heuristic. It does **not** connect to live ad platforms, campaign managers, or analytics dashboards. It relies on the user's provided spend notes, performance context, and channel mix.
## Trigger
Use this skill when the user wants to:
- review ad spend efficiency across multiple channels (Meta, Google, Amazon, TikTok, Xiaohongshu)
- diagnose why a channel or campaign is underperforming relative to spend
- rebalance budget across awareness, consideration, and conversion campaign types
- prepare a monthly or quarterly media budget review brief
- identify where to cut spend or where to scale based on ROAS or MER signals
### Example prompts
- "Our Meta Ads ROAS dropped this month — should we reallocate budget?"
- "Help me review and rebalance our Q1 ad spend across Amazon, Google, and TikTok"
- "Diagnose why our TikTok campaign is burning budget without conversions"
- "Create a budget rebalancing brief for a $50k monthly ad spend"
## Workflow
1. Capture the total budget, channel mix, campaign types, and performance signals.
2. Apply the efficiency framework to score each channel and campaign type.
3. Identify underperforming channels, audience segments, and campaign types.
4. Generate rebalancing recommendations with expected impact.
5. Return a markdown rebalancing brief.
## Inputs
The user can provide any mix of:
- total ad budget and channel breakdown: e.g., Meta 40%, Google 30%, Amazon 20%, TikTok 10%
- campaign type mix: awareness, consideration, conversion, retargeting
- performance signals: ROAS, MER, CPM, CPC, CPA, CTR, conversion rate by channel
- audience segment notes: demographic, interest, lookalike, retarget
- business context: seasonal window, product launch, clearance, brand campaign
- constraints: minimum spend requirements, creative constraints, platform policies
## Outputs
Return a markdown brief with:
- budget health summary (total spend, channel mix, overall efficiency)
- channel efficiency scorecard (ROAS/MER, CPM, CPC, CPA per channel)
- campaign type efficiency breakdown (awareness vs. conversion)
- audience segment performance notes
- rebalancing recommendations with specific reallocation percentages
- expected impact estimates and risk notes
- creative or landing-page considerations that may affect efficiency
## Safety
- No live ad platform, campaign manager, or analytics API access.
- Efficiency scores are directional unless complete spend and revenue data is provided.
- Do not claim guaranteed ROAS improvements or budget savings.
- Budget decisions remain human-approved; automated bid or budget changes are out of scope.
## Best-fit Scenarios
- SMB and mid-market teams managing $10k-$500k monthly ad budgets
- operators running multi-channel campaigns without a dedicated media buyer
- teams needing a regular budget review cadence without heavy BI tooling
## Not Ideal For
- real-time bid management, automated campaign optimization, or live spend control
- businesses with incomplete or inconsistent spend reporting
- highly complex attribution scenarios requiring multi-touch modeling
## Acceptance Criteria
- Return markdown text.
- Include channel efficiency scorecard and rebalancing recommendations.
- Make efficiency assumptions explicit when data is partial.
- Keep the brief practical for ecommerce operators and media buyers.
FILE:handler.py
#!/usr/bin/env python3
import sys
from typing import Any, Dict, List, Union
CHANNEL_RULES = {
'Meta Ads': ['meta', 'facebook', 'instagram', 'fb'],
'Google Ads': ['google', 'youtube', 'gdn'],
'Amazon Sponsored': ['amazon', 'ams', 'sponsored'],
'TikTok Ads': ['tiktok', 'douyin'],
'Xiaohongshu': ['xiaohongshu', 'rednote', 'xhs', 'xiaohongshu ads'],
}
CAMPAIGN_RULES = {
'Awareness': ['awareness', 'reach', 'branding', 'awareness', 'video', 'upper funnel'],
'Consideration': ['consideration', 'traffic', 'engagement', 'mid funnel', 'video views'],
'Conversion': ['conversion', 'sales', 'dra', 'direct response', 'lower funnel', 'purchase'],
'Retargeting': ['retarget', 'remarketing', 'rlsa', 'custom audience', 'past visitor'],
}
BudgetInput = Union[str, Dict[str, Any]]
def _score_rules(text: str, rules: Dict[str, List[str]]) -> Dict[str, int]:
return {name: sum(1 for kw in kws if kw in text) for name, kws in rules.items()}
def _join(items: List[str]) -> str:
return ', '.join(items) if items else 'Not specified'
class AdBudgetRebalancer:
def __init__(self, user_input: BudgetInput):
self.raw = user_input
self.text = self._normalize_input(user_input)
self.lower = self.text.lower()
self.channels = self._detect_channels()
self.campaign_types = self._detect_campaign_types()
self.concern = self._detect_concern()
def _normalize_input(self, user_input: BudgetInput) -> str:
if isinstance(user_input, dict):
chunks: List[str] = []
for key in ['budget', 'channels', 'campaign_types', 'performance', 'concern', 'notes']:
value = user_input.get(key)
if not value:
continue
if isinstance(value, list):
value = ', '.join(str(v) for v in value)
chunks.append(f'{key}: {value}')
return ' | '.join(chunks)
return str(user_input or '').strip()
def _detect_channels(self) -> List[str]:
matched = [name for name, kws in CHANNEL_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Meta Ads', 'Google Ads']
def _detect_campaign_types(self) -> List[str]:
matched = [name for name, kws in CAMPAIGN_RULES.items() if any(kw in self.lower for kw in kws)]
return matched or ['Awareness', 'Conversion']
def _detect_concern(self) -> str:
if any(kw in self.lower for kw in ['drop', 'decline', 'down', 'worse', 'worsening']):
return 'Performance Decline'
if any(kw in self.lower for kw in ['rebalance', 'reallocate', 'shift', 'redistribute']):
return 'Budget Reallocation'
if any(kw in self.lower for kw in ['review', 'monthly', 'quarterly', 'audit']):
return 'Periodic Review'
return 'Budget Diagnosis'
def _channel_scorecard(self) -> List[Dict[str, str]]:
default_roas = {'Meta Ads': 2.0, 'Google Ads': 3.0, 'Amazon Sponsored': 4.0, 'TikTok Ads': 1.5, 'Xiaohongshu': 1.2}
default_cpm = {'Meta Ads': 12, 'Google Ads': 8, 'Amazon Sponsored': 15, 'TikTok Ads': 6, 'Xiaohongshu': 20}
rows = []
for ch in self.channels:
roas = default_roas.get(ch, 2.0)
cpm = default_cpm.get(ch, 10)
efficiency = '🟢 Strong' if roas >= 3 else '🟡 Moderate' if roas >= 1.5 else '🔴 Weak'
recommendation = (
'Scale +15–25% if ROAS is above platform average'
if roas >= 3 else
'Test new creative and audience before scaling'
if roas >= 1.5 else
'Diagnose creative, landing page, or audience mismatch before increasing spend'
)
rows.append({
'channel': ch,
'roas': f'{roas}x',
'cpm': f'cpm',
'efficiency': efficiency,
'recommendation': recommendation,
})
return rows
def _campaign_breakdown(self) -> List[Dict[str, str]]:
awareness = '🟡 Moderate' if 'Awareness' in self.campaign_types else '⚪ Low'
consideration = '🟡 Moderate' if 'Consideration' in self.campaign_types else '⚪ Low'
conversion = '🟢 High' if 'Conversion' in self.campaign_types else '⚪ Low'
retargeting = '🟡 Moderate' if 'Retargeting' in self.campaign_types else '⚪ Low'
return [
{'type': 'Awareness', 'recommended_pct': '20–30%', 'current': awareness, 'note': 'Brand building and top-of-funnel reach'},
{'type': 'Consideration', 'recommended_pct': '20–30%', 'current': consideration, 'note': 'Mid-funnel engagement and traffic'},
{'type': 'Conversion', 'recommended_pct': '35–50%', 'current': conversion, 'note': 'Direct response and sales'},
{'type': 'Retargeting', 'recommended_pct': '15–20%', 'current': retargeting, 'note': 'Past visitors and cart abandoners'},
]
def _rebalance_recommendations(self) -> List[str]:
recs = [
'Identify the top 2 channels by ROAS and increase each by 15–20% from lower-performing channels.',
'Shift 10–15% of Awareness spend to Retargeting if new customer acquisition cost is rising.',
'Test a new creative variant in the weakest channel before reducing budget further.',
'Set a 7-day ROAS review checkpoint after any reallocation to confirm direction.',
]
if self.concern == 'Performance Decline':
recs.insert(0, 'Pause the declining channel for 48 hours, diagnose creative fatigue vs. audience exhaustion.')
recs.insert(1, 'Cross-check whether landing page conversion has changed, which would explain ROAS decline.')
return recs
def _creative_notes(self) -> List[str]:
return [
'Meta: Video creative with captions outperforms static by 20–40% in feed environments.',
'Google: RSA (Responsive Search Ads) with 8–10 headlines and 4 descriptions outperform expanded text ads.',
'Amazon: A+ content and lifestyle images boost Sponsored Brand conversion by 5–15%.',
'TikTok: Native, unpolished UGC-style content outperforms brand-produced video in feed.',
'Xiaohongshu: High-quality image + conversational copy with specific use-case scenarios drives engagement.',
]
def render(self) -> str:
lines: List[str] = []
lines.append('# Ad Budget Rebalancing Brief')
lines.append('')
lines.append(f'**Review mode:** {self.concern}')
lines.append(f'**Channels in scope:** {_join(self.channels)}')
lines.append(f'**Campaign types:** {_join(self.campaign_types)}')
lines.append('**Method note:** This is a heuristic budget brief. No live ad platform, campaign manager, or analytics API was accessed.')
lines.append('')
lines.append('## Budget Health Summary')
lines.append('- Start by confirming total monthly spend, channel mix percentages, and overall ROAS or MER.')
lines.append('- Treat ROAS targets as channel-specific: Amazon at 3x+ is different from TikTok at 1.5x.')
lines.append('- If overall ROAS is declining but individual channels look healthy, investigate audience overlap or attribution conflict.')
lines.append('')
lines.append('## Channel Efficiency Scorecard')
lines.append('| Channel | Est. ROAS | Est. CPM | Efficiency | Recommendation |')
lines.append('|---|---|---|---|---|')
for row in self._channel_scorecard():
lines.append(f'| {row["channel"]} | {row["roas"]} | {row["cpm"]} | {row["efficiency"]} | {row["recommendation"]} |')
lines.append('')
lines.append('## Campaign Type Mix')
lines.append('| Type | Recommended % | Your Mix | Note |')
lines.append('|---|---|---|---|---|')
for row in self._campaign_breakdown():
lines.append(f'| {row["type"]} | {row["recommended_pct"]} | {row["current"]} | {row["note"]} |')
lines.append('')
lines.append('## Rebalancing Recommendations')
for idx, rec in enumerate(self._rebalance_recommendations(), 1):
lines.append(f'{idx}. {rec}')
lines.append('')
lines.append('## Creative & Landing-Page Considerations')
for note in self._creative_notes():
lines.append(f'- {note}')
lines.append('')
lines.append('## Next Review Checkpoint')
lines.append('- Run a 7-day performance review after any budget reallocation.')
lines.append('- Set a monthly cadence for budget rebalancing even when performance is stable.')
lines.append('- Escalate to a full audit if ROAS drops more than 20% in any channel without clear creative or audience explanation.')
return '\n'.join(lines)
def handle(user_input: BudgetInput) -> str:
return AdBudgetRebalancer(user_input).render()
if __name__ == '__main__':
payload = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
print(handle(payload))
FILE:tests/test_handler.py
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from handler import AdBudgetRebalancer, handle
def test_channel_detection_meta():
rebalancer = AdBudgetRebalancer('Review our Meta Ads ROAS and reallocate budget')
assert 'Meta Ads' in rebalancer.channels
def test_channel_detection_google_tiktok():
rebalancer = AdBudgetRebalancer('Rebalance budget across Google and TikTok')
assert 'Google Ads' in rebalancer.channels
assert 'TikTok Ads' in rebalancer.channels
def test_campaign_type_detection():
rebalancer = AdBudgetRebalancer('Awareness campaigns on Meta and conversion campaigns on Google')
assert 'Awareness' in rebalancer.campaign_types
assert 'Conversion' in rebalancer.campaign_types
def test_concern_detection_performance_decline():
rebalancer = AdBudgetRebalancer('Our ROAS dropped this month on TikTok')
assert rebalancer.concern == 'Performance Decline'
def test_render_contains_sections():
output = handle('Our Meta Ads ROAS dropped this month — should we reallocate budget?')
assert output.startswith('# Ad Budget Rebalancing Brief')
assert '## Channel Efficiency Scorecard' in output
assert '## Rebalancing Recommendations' in output
def test_dict_input_supported():
output = handle({
'budget': '50k',
'channels': ['Meta Ads', 'Google Ads', 'Amazon Sponsored'],
'performance': {'Meta': 'ROAS 1.8', 'Google': 'ROAS 3.2'},
'concern': 'Periodic Review'
})
assert '# Ad Budget Rebalancing Brief' in output
assert 'Meta Ads' in output
if __name__ == '__main__':
for name, fn in list(globals().items()):
if name.startswith('test_') and callable(fn):
fn()
print('All tests passed.')