()
| 6 | from a3s_code import Agent |
| 7 | |
| 8 | def main(): |
| 9 | # Create agent from config |
| 10 | agent = Agent.create("~/.a3s/config.acl") |
| 11 | session = agent.session(".") |
| 12 | |
| 13 | print("Starting long-running operation...") |
| 14 | |
| 15 | # Cancel after 3 seconds |
| 16 | def cancel_after_delay(): |
| 17 | time.sleep(3) |
| 18 | print("\n🛑 Cancelling operation...") |
| 19 | cancelled = session.cancel() |
| 20 | print(f"Cancelled: {cancelled}") |
| 21 | |
| 22 | t = threading.Thread(target=cancel_after_delay) |
| 23 | t.start() |
| 24 | |
| 25 | # Start a long operation |
| 26 | try: |
| 27 | result = session.send("Write a very long story about a robot learning to code. Make it at least 5000 words.") |
| 28 | print("\n✅ Operation completed (possibly partial)") |
| 29 | print(f"Response length: {len(result.text)} chars") |
| 30 | print(f"First 200 chars: {result.text[:200]}") |
| 31 | except Exception as e: |
| 32 | print(f"\n❌ Operation failed: {e}") |
| 33 | |
| 34 | t.join() |
| 35 | |
| 36 | if __name__ == "__main__": |
| 37 | main() |
no test coverage detected